This is the full developer documentation for Forge # Forge backend infrastructure > Eight backend primitives with one versioned contract across Rust, JavaScript, Python, and Go. Forge runs inside an application backend. It provides key/value storage, queues, pub/sub, blobs, authentication helpers, rate limits, schedules, and configuration through one contract. PostgreSQL is the shared durable backend; memory is the process-local test backend. Blob bytes may also use a filesystem or S3-compatible store. The application owns its HTTP or RPC framework, business tables, authorization policy, worker handlers, deployment, process signals, and telemetry export. Frontend frameworks, browser transports, client caches, WebSockets, SSE clients, and refetching are also application concerns. Forge ships no browser package. ## Eight primitives, one connection [Section titled “Eight primitives, one connection”](#eight-primitives-one-connection) [Key/value](/primitives/#key-value)Get, set, counters, compare-and-swap, prefix scan, TTLs. Lineage: Redis. [Queue](/primitives/#queue)Background jobs with acks, retries, delays, dedup, and dead letters. Lineage: SQS. [Pub/sub](/primitives/#pubsub)Connected-only backend fan-out with no retention or replay. Lineage: Postgres LISTEN/NOTIFY. [Blob](/primitives/#blob)Put, get, delete, and presigned upload and download URLs. Lineage: S3. [Auth](/primitives/#auth)argon2 password hashing, sessions, and API keys. Lineage: OWASP and PHC. [Rate limit](/primitives/#rate-limit)Token-bucket or sliding-window checks with IETF headers. [Schedule](/primitives/#schedule)Recurring cron jobs and one-off work at a future time. [Config and flags](/primitives/#config-and-flags)Runtime settings and feature flags with percentage rollout. ## Runtime boundary [Section titled “Runtime boundary”](#runtime-boundary) One coordinated schema PostgreSQL-backed primitives use one Forge-owned schema and bounded pools. A measured hot primitive can use a separate PostgreSQL target without changing the public contract. One contract in four languages Rust, JavaScript, Python, and Go are checked against one versioned contract and the same declarative conformance scenarios. Explicit backend differences Memory and PostgreSQL implement the same supported operations, while runtime capabilities report durability, process scope, notification, and blob differences. Application-owned integration Forge does not install a web server, signal handler, metrics exporter, MCP server, or frontend adapter. The host application owns those boundaries. ## Minimal PostgreSQL setup [Section titled “Minimal PostgreSQL setup”](#minimal-postgresql-setup) forge.toml ```toml [postgres] url = "${DATABASE_URL:-postgres://localhost/myapp}" ``` `init()` reads `forge.toml` and returns one client. Development may run Forge migrations during initialization. Production startup validates the schema by default; run migrations as a separate deployment step before serving traffic. [Quickstart](/quickstart/)Install Forge and make your first calls in five minutes. [Tutorials](/tutorials/todo-app/)Build a to-do app, a URL shortener, and a chat app, step by step. [Primitives](/primitives/)The eight building blocks, with the methods that matter for each. [Semantics](/semantics/)Durability, ordering, retries, expiry, outages, and crash boundaries. # API stability > Forge's cross-language stability policy. Starting with 1.1.0, Rust, JavaScript, Python, and Go move in one coordinated release. A later 1.x release does not remove or rename a public API, change documented behavior or error classification, reinterpret configuration, or write a schema that the promised rollback version cannot use. Breaking changes wait for 2.0. Deprecations are announced in release notes and documentation. Where a warning is practical, the deprecated surface warns for at least one release. Removal happens only at the next declared compatibility boundary. The canonical machine-readable contract is `contract/forge.json`. Generated declarations, inventories, and reference tables come from it. Release CI rejects non-empty development gaps and compares the contract plus each language artifact against the compatibility baseline. Public Rust enums and options remain non-exhaustive. Queue receipts and cursors remain opaque. Applications must never parse either format. # Configuration > One explicit runtime profile, one PostgreSQL-centered production model, and one database-free test model. `forge.toml` uses the same runtime-profile keys and naming in Rust, JavaScript, Python, and Go. `init()` reads `./forge.toml`; `initFrom`/`init_from`/`InitFrom` reads another path; `initFromString`/`init_from_string`/`init_from_str`/`InitFromString` parses TOML already held in memory. Unknown keys and missing environment variables fail before Forge makes network calls. ## PostgreSQL profile [Section titled “PostgreSQL profile”](#postgresql-profile) PostgreSQL is the default and normal production profile. It owns Forge’s durable metadata, coordination, leases, schedules, authentication records, configuration, shared rate limits, and—unless another blob byte store is selected—blob bytes. forge.toml ```toml [forge] mode = "postgres" environment = "production" namespace = "myapp" [postgres] url = "${DATABASE_URL}" max_connections = 10 acquire_timeout_secs = 30 auto_migrate = false migration_lock_timeout_secs = 30 statement_timeout_ms = 15000 lock_timeout_ms = 5000 idle_in_transaction_timeout_ms = 15000 [queue] dedup_window_secs = 300 payload_retention_secs = 86400 terminal_retention_secs = 604800 succeeded_retention_secs = 604800 dead_retention_secs = 2592000 cancelled_retention_secs = 604800 [ratelimit] fail_open = true [blob] backend = "postgres" signing_secret = "${FORGE_BLOB_SIGNING_SECRET}" base_url = "/api/files" ``` Queue payload bytes and terminal status have separate retention clocks. Payload retention defaults to one day. Succeeded, dead, and cancelled status each default to seven days and can be tuned independently; `terminal_retention_secs` sets all three before an individual override. Normal initialization requires a reachable PostgreSQL service, usable credentials, and a compatible Forge schema. Development and test default `auto_migrate` to true; production defaults it to false. With automatic migration disabled, initialization performs a read-only schema validation and tells the operator to run the migration API when the schema is pending, locked, incompatible, or drifted. Applications use whichever primitives they need; there is no feature-enable matrix and no configuration-driven plugin loader. Forge requires PostgreSQL 18. The system pool defaults to ten connections and requires at least two while startup migrations are enabled; validation-only initialization can use one. `acquire_timeout_secs` and `migration_lock_timeout_secs` are seconds. The three server-side timeouts are milliseconds and accept `0` to use PostgreSQL’s unlimited default. ### Embedded PostgreSQL [Section titled “Embedded PostgreSQL”](#embedded-postgresql) Local Rust, Node, and Python development can run a real embedded PostgreSQL server: ```toml [forge] mode = "postgres" environment = "development" [postgres] url = "${DATABASE_URL:-}" embedded = true embedded_dir = ".forge/pg" ``` A non-empty URL wins; otherwise Forge starts the embedded server. The Rust package requires the `embedded` feature. Go intentionally uses the system PostgreSQL profile or explicit memory profile and does not download an embedded server. Embedded state is durable on that machine but is not a multi-replica production deployment. ## Database-free memory profile [Section titled “Database-free memory profile”](#database-free-memory-profile) Memory is one complete runtime variant, not eight independent backend switches: ```toml [forge] mode = "memory" environment = "test" namespace = "unit-test" ``` Memory initialization creates no pool, starts no embedded service, runs no migration, and requires no PostgreSQL URL. Every primitive keeps the same public API, so tests do not branch around Forge. A declared production environment rejects memory unless `allow_memory_in_production = true` is written explicitly. | Property | PostgreSQL | Memory | | ------------------- | ------------------------------------------------------- | --------------------------- | | Restart persistence | Durable | Lost on process exit | | Sharing | Processes and replicas | One process | | Pub/sub | Cross-process LISTEN/NOTIFY, connected subscribers only | In-process subscribers only | | Rate-limit scope | Shared | Process-local | | PostgreSQL accessor | Pool/credential-bearing URL | `NOT_CONFIGURED` | `backendCapabilities()` / `backend_capabilities()` / `BackendCapabilities()` provides one provider line per primitive without performing a health check. Use `probe()` / `Probe()` for bounded live readiness checks. Caution Memory mode is not a production cache tier and does not silently fall back from PostgreSQL. A PostgreSQL outage makes initialization fail. ## Blob bytes [Section titled “Blob bytes”](#blob-bytes) `blob.backend = "postgres"` keeps bytes and metadata in PostgreSQL, which preserves the default “Postgres is enough” deployment. `blob.backend = "fs"` stores bytes below `blob.fs_root` while PostgreSQL retains Forge’s durable metadata; use it only with a shared mount when multiple replicas must see the same objects. The explicit memory profile selects memory blob storage automatically. S3-compatible storage is the intentional production backend for large bytes, but it does not replace Forge’s system PostgreSQL responsibilities. forge.toml · S3-compatible blob bytes ```toml [blob] backend = "s3" bucket = "my-app-production" region = "us-east-1" prefix = "forge" path_style = false connect_timeout_secs = 3 request_timeout_secs = 30 max_retries = 3 signing_secret = "${FORGE_BLOB_SIGNING_SECRET}" # Omit endpoint for AWS S3. Set it, and usually path_style = true, for MinIO/R2/etc. # endpoint = "https://object-storage.example.com" # Omit static credentials to use the standard AWS credential chain. # access_key = "${S3_ACCESS_KEY}" # secret_key = "${S3_SECRET_KEY}" # session_token = "${S3_SESSION_TOKEN}" ``` Initialization performs `HeadBucket` followed by create-and-abort multipart operations. This verifies bucket access and required permissions without leaving a permanent object. Access keys, secret keys, session tokens, and presigned URL query strings are credentials and must remain out of logs and traces. ## Measured database isolation [Section titled “Measured database isolation”](#measured-database-isolation) A hot primitive can use a separate PostgreSQL target without changing its schema version or public behavior: ```toml [databases.queue] url = "${QUEUE_DATABASE_URL}" max_connections = 20 ``` Every distinct target receives the same coordinated Forge migration history. This is a measured pool/server bulkhead, not an independently versioned primitive database. ## Namespaces and environment interpolation [Section titled “Namespaces and environment interpolation”](#namespaces-and-environment-interpolation) `forge.namespace` scopes all keys, queues, rows, topics, schedules, caches, notifications, signing operations, maintenance work, and authentication records. It is case-sensitive, immutable after `init`, not normalized, and may contain at most 128 bytes using ASCII letters, digits, `.`, `_`, and `-`. Empty means the unnamespaced profile and must not share a backend with an application that expects isolation. Namespacing prevents accidental collisions; it does not authenticate a caller or authorize access. Applications make authorization decisions before calling Forge. The `scopeKvKey` / `scope_kv_key` / `ScopeKVKey` family composes bounded application, tenant, user, and resource names separately for KV keys, blob keys, rate-limit subjects, and topics without installing a hidden universal tenant context. ### Application scopes [Section titled “Application scopes”](#application-scopes) The v1 helpers require all four components in one call. They validate 1–255 UTF-8 bytes, reject control characters, and check the selected primitive’s complete backend-safe limit before any Forge backend call. Length-prefixed UTF-8 components make delimiters unambiguous, and `parseScopedName` / `parse_scoped_name` / `ParseScopedName` recovers the four components. ```rust let key = forgelib::scope_kv_key("billing", "tenant-7", "user-42", "invoice:9")?; let parsed = forgelib::parse_scoped_name(&key)?; ``` ```ts import { scopeKvKey, parseScopedName } from "forgelib"; const key = scopeKvKey("billing", "tenant-7", "user-42", "invoice:9"); const parsed = parseScopedName(key); ``` ```python key = forgelib.scope_kv_key("billing", "tenant-7", "user-42", "invoice:9") parsed = forgelib.parse_scoped_name(key) ``` ```go key, err := forge.ScopeKVKey("billing", "tenant-7", "user-42", "invoice:9") parsed, err := forge.ParseScopedName(key) ``` These helpers produce names, not permissions. Authorize `tenant-7`, `user-42`, and the invoice in application code before calling Forge; parsing a key proves only that it is well formed. Queue names, authentication identities, config keys, schedules, and other primitives remain independently named because one hidden tenant prefix would create incorrect authorization assumptions. String values accept `${VAR}` and `${VAR:-default}`; a missing variable without a default is an error, and interpolation cannot inject TOML structure. Environment interpolation and `FORGE_CFG_*` config overrides are captured deterministically when the process initializes. Changing the host environment afterward does not mutate a running Forge instance; write dynamic values through the config API instead. Never log `postgresUrl()`/`postgres_url()`/`PostgresURL()`: it returns a credential-bearing DSN. In memory mode the same accessor returns `NOT_CONFIGURED`. ## Lifecycle [Section titled “Lifecycle”](#lifecycle) Applications own process signals. On shutdown, stop accepting requests and call `close` with a deadline. Close is idempotent, stops new work, wakes subscriptions, closes dedicated listeners and pools, and stops an embedded server. Native Rust and Go workers stop dequeuing and drain handlers within their configured grace; after grace they cancel or abort supervisors and release or expire leases. JavaScript and Python userland workers cancel active handlers and release leases immediately; JavaScript handlers can inspect `job.signal`, while Python handlers receive task cancellation and can inspect `job.cancelled`. Forge never installs SIGINT or SIGTERM handlers. Rust’s trait-injection builder remains an advanced compile-time extension for custom backends. It is not exposed as portable configuration, dynamic discovery, or a plugin loader; new first-party backends require evidence and a complete conformance story. # Contract reference > Generated cross-language operation and result mappings. Contract version: `1.1.0`. Durations use seconds unless a field ends in `_ms`; sizes use bytes. | Operation | Rust | JavaScript | Python | Go | Result | | ------------------------------------- | --------------------------------------------- | ------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------ | --------------------------- | ----------------------- | | `client.init` | `init` | `init` | `init` | `InitDefault` | `Forge` | | `client.init_from` | `init_from` | `initFrom` | `init_from` | `InitFrom` | `Forge` | | `client.init_from_string` | `init_from_str` | `initFromString` | `init_from_string` | `InitFromString` | `Forge` | | `client.init_memory_for_testing` | `init_memory_for_testing` | `initMemoryForTesting` | `init_memory_for_testing` | `NewMemoryForTesting` | `Forge` | | `client.advance_test_clock` | `advance_test_clock` | `advanceTestClock` | `advance_test_clock` | `AdvanceTestClock` | `void` | | `client.migrate` | `migrate` | `migrate` | `migrate` | `Migrate` | `MigrationReport[]` | | `client.migrate_from` | `migrate_from` | `migrateFrom` | `migrate_from` | `MigrateFrom` | `MigrationReport[]` | | `client.migrate_from_string` | `migrate_from_str` | `migrateFromString` | `migrate_from_string` | `MigrateFromString` | `MigrationReport[]` | | `client.migration_status` | `migration_status` | `migrationStatus` | `migration_status` | `MigrationStatus` | `MigrationReport[]` | | `client.migration_status_from` | `migration_status_from` | `migrationStatusFrom` | `migration_status_from` | `MigrationStatusFrom` | `MigrationReport[]` | | `client.migration_status_from_string` | `migration_status_from_str` | `migrationStatusFromString` | `migration_status_from_string` | `MigrationStatusFromString` | `MigrationReport[]` | | `client.validate_schema` | `validate_schema` | `validateSchema` | `validate_schema` | `ValidateSchema` | `MigrationReport[]` | | `client.validate_schema_from` | `validate_schema_from` | `validateSchemaFrom` | `validate_schema_from` | `ValidateSchemaFrom` | `MigrationReport[]` | | `client.validate_schema_from_string` | `validate_schema_from_str` | `validateSchemaFromString` | `validate_schema_from_string` | `ValidateSchemaFromString` | `MigrationReport[]` | | `client.close` | `close` | `close` | `close` | `Close` | `void` | | `client.backend_capabilities` | `backend_capabilities` | `backendCapabilities` | `backend_capabilities` | `BackendCapabilities` | `BackendInfo[]` | | `client.liveness` | `is_live` | `isLive` | `is_live` | `IsLive` | `boolean` | | `client.probe` | `probe` | `probe` | `probe` | `Probe` | `HealthReport` | | `client.metrics_snapshot` | `metrics_snapshot` | `metricsSnapshot` | `metrics_snapshot` | `MetricsSnapshot` | `MetricSample[]` | | `client.diagnostics` | `diagnostics` | `diagnostics` | `diagnostics` | `Diagnostics` | `DiagnosticsReport` | | `client.render_prometheus` | `render_prometheus` | `renderPrometheus` | `render_prometheus` | `RenderPrometheus` | `string` | | `client.maintain` | `maintain` | `maintain` | `maintain` | `Maintain` | `void` | | `client.postgres_url` | `postgres_url` | `postgresUrl` | `postgres_url` | `PostgresURL` | `string` | | `kv.get` | `get` | `kvGet`, `kvGetBytes` | `kv_get`, `kv_get_bytes` | `KVGet` | `bytes?` | | `kv.mget` | `mget` | `kvMget` | `kv_mget` | `KVMGet` | `bytes?[]` | | `kv.set` | `set` | `kvSet`, `kvSetBytes` | `kv_set`, `kv_set_bytes` | `KVSet` | `boolean` | | `kv.delete` | `delete` | `kvDelete` | `kv_delete` | `KVDelete` | `boolean` | | `kv.exists` | `exists` | `kvExists` | `kv_exists` | `KVExists` | `boolean` | | `kv.incr` | `incr` | `kvIncr` | `kv_incr` | `KVIncr` | `int64` | | `kv.expire` | `expire` | `kvExpire` | `kv_expire` | `KVExpire` | `boolean` | | `kv.compare_and_swap` | `compare_and_swap` | `kvCompareAndSwap` | `kv_compare_and_swap` | `KVCompareAndSwap` | `boolean` | | `kv.scan` | `scan` | `kvScan`, `kvScanPage` | `kv_scan`, `kv_scan_page` | `KVScan` | `ScanPage` | | `queue.enqueue` | `enqueue` | `queueEnqueue` | `queue_enqueue` | `Enqueue` | `JobId` | | `queue.dequeue` | `dequeue` | `queueDequeue` | `queue_dequeue` | `Dequeue` | `Job?` | | `queue.enqueue_batch` | `enqueue_batch` | `queueEnqueueBatch` | `queue_enqueue_batch` | `EnqueueBatch` | `BatchEnqueueResult[]` | | `queue.dequeue_batch` | `dequeue_batch` | `queueDequeueBatch` | `queue_dequeue_batch` | `DequeueBatch` | `Job[]` | | `queue.ack` | `ack` | `queueAck` | `queue_ack` | `Ack` | `void` | | `queue.nack` | `nack` | `queueNack` | `queue_nack` | `Nack` | `void` | | `queue.heartbeat` | `heartbeat` | `queueHeartbeat` | `queue_heartbeat` | `Heartbeat` | `void` | | `queue.cancellation_requested` | `cancellation_requested` | `queueCancellationRequested` | `queue_cancellation_requested` | `CancellationRequested` | `boolean` | | `queue.finish_cancellation` | `finish_cancellation` | `queueFinishCancellation` | `queue_finish_cancellation` | `FinishCancellation` | `void` | | `queue.cancel` | `cancel` | `queueCancel` | `queue_cancel` | `CancelJob` | `JobStatus?` | | `queue.status` | `status` | `queueStatus` | `queue_status` | `JobStatus` | `JobStatus?` | | `queue.list_status` | `list_status` | `queueListStatus` | `queue_list_status` | `ListJobStatus` | `JobStatusPage` | | `queue.depth` | `depth` | `queueDepth` | `queue_depth` | `Depth` | `QueueDepth` | | `queue.dead_letters` | `dead_letters` | `queueDeadLetters` | `queue_dead_letters` | `DeadLetters` | `DeadLetterPage` | | `queue.pause` | `pause` | `queuePause` | `queue_pause` | `PauseQueue` | `void` | | `queue.resume` | `resume` | `queueResume` | `queue_resume` | `ResumeQueue` | `void` | | `queue.is_paused` | `is_paused` | `queueIsPaused` | `queue_is_paused` | `QueuePaused` | `boolean` | | `queue.stats` | `stats` | `queueStats` | `queue_stats` | `QueueStats` | `QueueStats` | | `queue.redrive` | `redrive` | `queueRedrive` | `queue_redrive` | `Redrive` | `boolean` | | `queue.redrive_batch` | `redrive_batch` | `queueRedriveBatch` | `queue_redrive_batch` | `RedriveBatch` | `RedriveBatchResult` | | `queue.purge_dead_letters_dry_run` | `purge_dead_letters_dry_run` | `queuePurgeDeadLettersDryRun` | `queue_purge_dead_letters_dry_run` | `PurgeDeadLettersDryRun` | `count` | | `queue.purge_dead_letters` | `purge_dead_letters` | `queuePurgeDeadLetters` | `queue_purge_dead_letters` | `PurgeDeadLetters` | `count` | | `queue.outbox_once` | `run_outbox_once` | `runOutboxOnce` | `run_outbox_once` | `RunOutboxOnce` | `OutboxRelayReport` | | `queue.outbox_relay` | `run_outbox_relay`, `run_outbox_relay_until` | `runOutboxRelay` | `run_outbox_relay` | `RunOutboxRelay` | `unit` | | `config.get_raw` | `get_raw` | `configGet` | `config_get` | `ConfigGet` | `bytes?` | | `config.set_raw` | `set_raw` | `configSet` | `config_set` | `ConfigSet` | `void` | | `config.get_many_raw` | `get_many_raw` | `configGetMany` | `config_get_many` | `ConfigGetMany` | `ConfigEntry[]` | | `config.delete_raw` | `delete_raw` | `configDelete` | `config_delete` | `ConfigDelete` | `boolean` | | `config.flag` | `flag` | `flag` | `flag` | `Flag` | `boolean` | | `config.set_flag` | `set_flag` | `setFlagPercent`, `setFlagOn`, `setFlagOff`, `setFlagAllowList`, `setFlagValue` | `set_flag_percent`, `set_flag_on`, `set_flag_off`, `set_flag_allow_list`, `set_flag_value` | `SetFlag` | `void` | | `config.flag_details` | `flag_details` | `flagDetails` | `flag_details` | `FlagDetails` | `FlagEvaluation` | | `config.delete_flag` | `delete_flag` | `deleteFlag` | `delete_flag` | `DeleteFlag` | `boolean` | | `config.flag_details_many` | `flag_details_many` | `flagDetailsMany` | `flag_details_many` | `FlagDetailsMany` | `FlagEvaluationEntry[]` | | `config.snapshot` | `snapshot` | `configSnapshot` | `config_snapshot` | `ConfigSnapshot` | `ConfigSnapshot` | | `config.snapshot_encode` | `encode_snapshot` | `encodeConfigSnapshot` | `encode_config_snapshot` | `EncodeConfigSnapshot` | `bytes` | | `config.snapshot_decode` | `decode_snapshot` | `decodeConfigSnapshot` | `decode_config_snapshot` | `DecodeConfigSnapshot` | `ConfigSnapshot` | | `ratelimit.check` | `check`, `check_with` | `rateLimitCheck` | `rate_limit_check` | `RateLimitCheck` | `Decision` | | `ratelimit.check_weighted` | `check_cost`, `check_cost_with` | `rateLimitCheck` | `rate_limit_check` | `RateLimitCheck` | `Decision` | | `ratelimit.reserve` | `reserve` | `rateLimitReserve` | `rate_limit_reserve` | `RateLimitReserve` | `Reservation?` | | `ratelimit.commit` | `commit` | `rateLimitCommit` | `rate_limit_commit` | `RateLimitCommit` | `Reservation` | | `ratelimit.release` | `release` | `rateLimitRelease` | `rate_limit_release` | `RateLimitRelease` | `Reservation` | | `blob.put` | `put` | `blobPut`, `blobPutBytes`, `blobPutObject` | `blob_put`, `blob_put_object` | `BlobPut` | `void` | | `blob.put_stream` | `put_stream` | `blobPutFile` | `blob_put_file` | `BlobPutStream` | `void` | | `blob.get` | `get`, `open` | `blobGet`, `blobGetBytes` | `blob_get` | `BlobGet`, `BlobOpen` | `bytes?` | | `blob.get_range` | `get_range` | `blobGetRange` | `blob_get_range` | `BlobGetRange` | `bytes?` | | `blob.get_if` | `get_if` | `blobGetIf` | `blob_get_if` | `BlobGetIf` | `ConditionalBlobGet` | | `blob.head` | `head` | `blobHead`, `blobContentType` | `blob_head`, `blob_content_type` | `BlobHead` | `BlobInfo?` | | `blob.delete` | `delete` | `blobDelete` | `blob_delete` | `BlobDelete` | `void` | | `blob.list` | `list` | `blobList` | `blob_list` | `BlobList` | `BlobPage` | | `blob.copy` | `copy` | `blobCopy` | `blob_copy` | `BlobCopy` | `BlobInfo` | | `blob.create_multipart` | `create_multipart` | `blobCreateMultipart` | `blob_create_multipart` | `BlobCreateMultipart` | `MultipartUpload` | | `blob.upload_part` | `upload_part` | `blobUploadPart` | `blob_upload_part` | `BlobUploadPart` | `MultipartPart` | | `blob.complete_multipart` | `complete_multipart` | `blobCompleteMultipart` | `blob_complete_multipart` | `BlobCompleteMultipart` | `BlobInfo` | | `blob.abort_multipart` | `abort_multipart` | `blobAbortMultipart` | `blob_abort_multipart` | `BlobAbortMultipart` | `void` | | `blob.verify_checksum_sha256` | `verify_checksum_sha256` | `blobVerifyChecksumSha256` | `blob_verify_checksum_sha256` | `BlobVerifyChecksumSHA256` | `boolean` | | `blob.presign_upload` | `presign_upload` | `blobPresignUpload` | `blob_presign_upload` | `BlobPresignUpload` | `ProxyPresign` | | `blob.presign_download` | `presign_download` | `blobPresignDownload` | `blob_presign_download` | `BlobPresignDownload` | `ProxyPresign` | | `blob.presign_native_get` | `presign_native_get` | `blobPresignNativeGet` | `blob_presign_native_get` | `BlobPresignNativeGet` | `NativePresign` | | `blob.presign_native_put` | `presign_native_put` | `blobPresignNativePut` | `blob_presign_native_put` | `BlobPresignNativePut` | `NativePresign` | | `blob.verify_presigned` | `verify_presigned` | `blobVerifyPresign` | `blob_verify_presign` | `BlobVerifyPresigned` | `boolean` | | `auth.hash_password` | `hash_password` | `hashPassword` | `hash_password` | `HashPassword` | `string` | | `auth.verify_password` | `verify_password` | `verifyPassword` | `verify_password` | `VerifyPassword` | `boolean` | | `auth.needs_rehash` | `needs_rehash` | `needsRehash` | `needs_rehash` | `NeedsRehash` | `boolean` | | `auth.create_session` | `create_session` | `createSession` | `create_session` | `CreateSession` | `string` | | `auth.validate_session` | `validate_session` | `validateSession`, `validateSessionInfo` | `validate_session`, `validate_session_info` | `ValidateSession` | `Session?` | | `auth.revoke_session` | `revoke_session` | `revokeSession` | `revoke_session` | `RevokeSession` | `void` | | `auth.revoke_all_sessions` | `revoke_all_sessions` | `revokeAllSessions` | `revoke_all_sessions` | `RevokeAllSessions` | `uint64` | | `auth.create_api_key` | `create_api_key`, `create_api_key_with` | `createApiKey`, `createApiKeyWith` | `create_api_key`, `create_api_key_with` | `CreateAPIKey` | `ApiKey` | | `auth.verify_api_key` | `verify_api_key` | `verifyApiKey` | `verify_api_key` | `VerifyAPIKey` | `ApiKeyInfo?` | | `auth.revoke_api_key` | `revoke_api_key` | `revokeApiKey` | `revoke_api_key` | `RevokeAPIKey` | `void` | | `auth.create_token` | `create_token`, `create_token_with_payload` | `createToken`, `createTokenWithPayload` | `create_token`, `create_token_with_payload` | `CreateToken` | `string` | | `auth.consume_token` | `consume_token`, `consume_token_with_payload` | `consumeToken`, `consumeTokenWithPayload` | `consume_token`, `consume_token_with_payload` | `ConsumeToken` | `TokenConsumption?` | | `scope.kv_key` | `scope_kv_key` | `scopeKvKey` | `scope_kv_key` | `ScopeKVKey` | `string` | | `scope.blob_key` | `scope_blob_key` | `scopeBlobKey` | `scope_blob_key` | `ScopeBlobKey` | `string` | | `scope.rate_limit_subject` | `scope_rate_limit_subject` | `scopeRateLimitSubject` | `scope_rate_limit_subject` | `ScopeRateLimitSubject` | `string` | | `scope.topic` | `scope_topic` | `scopeTopic` | `scope_topic` | `ScopeTopic` | `string` | | `schedule.cron` | `cron` | `scheduleCron` | `schedule_cron` | `ScheduleCron` | `void` | | `schedule.at` | `at` | `scheduleAt` | `schedule_at` | `ScheduleAt` | `JobId` | | `schedule.cancel` | `cancel` | `scheduleCancel` | `schedule_cancel` | `ScheduleCancel` | `boolean` | | `schedule.cancel_at` | `cancel_at` | `scheduleCancelAt` | `schedule_cancel_at` | `ScheduleCancelAt` | `boolean` | | `schedule.inspect` | `inspect` | `scheduleInspect` | `schedule_inspect` | `ScheduleInspect` | `ScheduleInfo?` | | `schedule.pause` | `pause` | `schedulePause` | `schedule_pause` | `SchedulePause` | `boolean` | | `schedule.resume` | `resume` | `scheduleResume` | `schedule_resume` | `ScheduleResume` | `boolean` | | `schedule.diagnostics` | `diagnostics` | `schedulerDiagnostics` | `scheduler_diagnostics` | `SchedulerDiagnostics` | `SchedulerDiagnostics` | | `schedule.list` | `list` | `scheduleList` | `schedule_list` | `ScheduleList` | `SchedulePage` | | `schedule.process_due` | `process_due` | `runSchedulerOnce` | `run_scheduler_once` | `RunSchedulerOnce` | `uint64` | | `pubsub.channel_for` | `channel_for` | `pubsubChannel` | `pubsub_channel` | `PubsubChannel` | `string` | | `pubsub.publish` | `publish` | `pubsubPublish` | `pubsub_publish` | `Publish` | `void` | | `pubsub.subscribe` | `subscribe` | `pubsubSubscribe` | `pubsub_subscribe` | `Subscribe` | `Subscription` | # Event delivery > Publish bounded invalidation hints from Forge and deliver them through application-owned transports. Forge is a backend library. It can encode a bounded invalidation hint and publish it from the server, but it does not own an application’s HTTP routes, WebSockets, server-sent events, browser cache, or client-side refetching. The usual flow is simple. The backend commits authoritative state, publishes a hint, and an application-owned transport forwards that hint to connected clients. A client that cares about the hint fetches authoritative state again. Loss, duplication, delay, and reordering must not change correctness. ## Invalidation event v1 [Section titled “Invalidation event v1”](#invalidation-event-v1) The canonical schema is [`contract/invalidation-v1.schema.json`](https://github.com/isala404/forge/blob/main/contract/invalidation-v1.schema.json). A version-1 event has application-defined `tags`, application-defined `query_keys` fragments, an optional opaque `revision`, and no authoritative state payload. ```json { "schema_version": 1, "tags": ["links/user-42"], "query_keys": [["links", { "owner": "user-42" }]], "revision": "184467" } ``` The encoded event is at most 4096 bytes. It has at most 32 unique tags, 32 query-key fragments, and 64 total targets. Tags are at most 128 UTF-8 bytes. A fragment has 1 to 8 parts, at most three nested JSON levels, 16 items per container, and 32 total nodes. A revision is at most 256 UTF-8 bytes. Version-1 decoders enforce the total byte limit and ignore unknown additive fields. They reject unknown schema versions. Use tags for coarse invalidation. A tag such as `invoices/account-42` is an application identifier, not a cache patch instruction. A query-key fragment is also only an input to application code. Forge does not infer client queries from URLs, response payloads, or database records. Caution An invalidation event is not a change event. It carries no row, patch, deletion, authorization decision, or executable instruction. Re-authorize every read that follows it. ## Publish after durable commit [Section titled “Publish after durable commit”](#publish-after-durable-commit) Each server runtime exposes the same encoder. Publish the encoded hint only after the authoritative transaction commits. Forge pub/sub reaches connected subscribers at most once and has no replay cursor, so a reconnect begins with an authoritative read. * Node ```ts import { encodeInvalidationEvent } from "forgelib"; await forge.pubsubPublish("account:42", encodeInvalidationEvent({ schema_version: 1, tags: ["invoices/account-42"], query_keys: [["invoices", { account: "42" }]], revision: committedRevision, })); ``` * Python ```python from forgelib import encode_invalidation_event await forge.pubsub_publish("account:42", encode_invalidation_event({ "schema_version": 1, "tags": ["invoices/account-42"], "query_keys": [["invoices", {"account": "42"}]], "revision": committed_revision, })) ``` * Rust ```rust use forgelib::{Bytes, InvalidationEvent}; use serde_json::json; let event = InvalidationEvent::new( vec!["invoices/account-42".into()], vec![vec![json!("invoices"), json!({ "account": "42" })]], Some(committed_revision), )?; forge.pubsub().publish("account:42", Bytes::from(event.encode()?)).await?; ``` * Go ```go revision := committedRevision payload, err := forge.EncodeInvalidationEvent(forge.InvalidationEvent{ SchemaVersion: 1, Tags: []string{"invoices/account-42"}, QueryKeys: [][]any{{"invoices", map[string]any{"account": "42"}}}, Revision: &revision, }) if err != nil { return err } return client.Publish(ctx, "account:42", payload) ``` ## Connect an application transport [Section titled “Connect an application transport”](#connect-an-application-transport) Keep the transport route inside Hono, Axum, FastAPI, Django, Go `net/http`, or the application’s existing framework. Before opening a subscription, authenticate the session, authorize the allowed topic, verify the request origin, and reserve one of a bounded number of connections for that principal. Do not accept an arbitrary Forge topic from a path or query parameter. The adapter should have a fixed topic allowlist, global and per-principal connection limits, heartbeats with write deadlines, and a bounded outbound buffer. Close slow consumers instead of accumulating messages. Release the Forge subscription and the connection slot on disconnect. Routing, cookies, CSRF protection, CORS, serialization, proxy buffering, idle timeouts, and deployment behavior remain application-owned. ```ts // Hono SSE shape. authenticateAndAuthorize returns a server-selected topic. const subscription = await limits.open(c.req.raw, async () => { checkOrigin(c.req.header("Origin")); const principal = await authenticate(c.req.raw); return authorizeInvalidationTopic(principal, c.req.param("resource")); }); return streamSSE(c, async (stream) => { const events = await forge.topic(subscription.topic).subscribe(); stream.onAbort(() => { void events.return?.(); subscription.release(); }); for await (const raw of withHeartbeatAndTimeout(events, 15_000)) { await writeWithDeadline(stream, { event: "forge-invalidation-v1", data: JSON.stringify(raw) }); } }); ``` ```python # FastAPI WebSocket shape. Accept only after every application check passes. check_origin(websocket.headers.get("origin")) principal = await authenticate(websocket) topic = authorize_invalidation_topic(principal, websocket.path_params["resource"]) slot = connection_limits.reserve(principal.id) await websocket.accept() try: async for raw in bounded_with_heartbeat(forge.topic(topic).subscribe(), max_buffer=32): await asyncio.wait_for(websocket.send_json(raw), timeout=5) finally: slot.release() ``` The same ownership applies to Axum WebSocket and Go HTTP/SSE routes. Extract the authenticated principal in middleware, map an allowed resource to a server-selected topic, reserve a slot, stream through a bounded channel, time out writes and heartbeats, and clean up on cancellation. ## Client ownership [Section titled “Client ownership”](#client-ownership) Applications choose their frontend framework, transport client, cache, and refetch policy. This includes React, Vue, TanStack Query, native mobile clients, WebSocket reconnect behavior, SSE retry behavior, optimistic updates, offline persistence, and browser cache lifetime. Forge ships no client adapter for these choices. On initial connection and every reconnect, fetch authoritative state before trusting later hints. Map tags and `query_keys` fragments only inside application code. Keep the fan-out bounded, ignore unknown targets, and re-authorize each fetch. A client cache may collapse duplicate hints or compare opaque revisions, but it must remain correct if it receives no hint at all. # Integrations > Framework, deployment, MCP, CloudEvents, environment, and diagnostics recipes that keep protocol ownership in the application. Forge is a library inside the process you already own. A framework recipe should inject one initialized handle, reject traffic until initialization succeeds, stop accepting traffic before closing Forge, and leave routing, authentication, protocol behavior, and process signals with the host. None of the recipes below add a framework dependency to a Forge core package. ## HTTP framework lifecycle [Section titled “HTTP framework lifecycle”](#http-framework-lifecycle) The canonical applications exercise Rocket, GraphQL Yoga, and FastAPI against real PostgreSQL. These smaller recipes show the ownership seam for other host styles. ### Axum [Section titled “Axum”](#axum) Put the cheap-to-clone `Forge` handle in router state. Axum owns the listener and shutdown future; Forge only drains its own workers and closes pools after the server stops accepting requests. ```rust #[derive(Clone)] struct AppState { forge: forgelib::Forge } let forge = forgelib::Forge::init().await?; let app = axum::Router::new() .route("/healthz", axum::routing::get(|| async { "ok" })) .with_state(AppState { forge: forge.clone() }); let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await?; axum::serve(listener, app).with_graceful_shutdown(shutdown_signal()).await?; forge.close(std::time::Duration::from_secs(30)).await?; ``` ### Hono [Section titled “Hono”](#hono) Create Forge before `serve`, attach it through Hono variables or an application closure, and close the Node server before the Forge handle. Install the signal handler in the executable, never in a shared package. ```ts const forge = await ForgeClient.init(); const app = new Hono<{ Variables: { forge: ForgeClient } }>(); app.use("*", async (c, next) => { c.set("forge", forge); await next(); }); const server = serve({ fetch: app.fetch, port: 3000 }); process.once("SIGTERM", () => { server.close(async () => { await forge.close(30); process.exit(0); }); }); ``` ### Express [Section titled “Express”](#express) Express request augmentation is application code. A closure is often simpler than changing global request types. ```ts const forge = await ForgeClient.init(); const app = express(); app.get("/profile/:id", async (req, res, next) => { try { res.json(await forge.kv(`profile:${req.params.id}`).get()); } catch (error) { next(error); } }); const server = app.listen(3000); process.once("SIGTERM", () => server.close(async () => { await forge.close(30); process.exit(0); })); ``` ### FastAPI [Section titled “FastAPI”](#fastapi) FastAPI lifespan is the single initialization and shutdown boundary. Store the handle on `app.state`; dependencies may narrow it for handlers. ```python @asynccontextmanager async def lifespan(app: FastAPI): forge = await forgelib.ForgeClient.init() app.state.forge = forge try: yield finally: await forge.close(30.0) app = FastAPI(lifespan=lifespan) def get_forge(request: Request) -> forgelib.ForgeClient: return request.app.state.forge ``` ### Django ASGI [Section titled “Django ASGI”](#django-asgi) Django’s request object does not own application lifespan. Wrap its ASGI callable once, initialize lazily under a lock for hosts that do not send lifespan events, and let a lifespan-aware host close the handle. Do not initialize a new pool in middleware per request. ```python django_app = get_asgi_application() class ForgeASGI: def __init__(self, app): self.app, self.forge, self.lock = app, None, asyncio.Lock() async def ready(self): async with self.lock: if self.forge is None: self.forge = await forgelib.ForgeClient.init() return self.forge async def __call__(self, scope, receive, send): if scope["type"] == "lifespan": while True: message = await receive() if message["type"] == "lifespan.startup": await self.ready(); await send({"type": "lifespan.startup.complete"}) elif message["type"] == "lifespan.shutdown": if self.forge is not None: await self.forge.close(30.0) await send({"type": "lifespan.shutdown.complete"}); return scope["forge"] = await self.ready() await self.app(scope, receive, send) application = ForgeASGI(django_app) ``` ### Go `net/http` [Section titled “Go net/http”](#go-nethttp) Construct one handle, close the HTTP server with a deadline, then close Forge with the remaining process context. ```go client, err := forge.InitDefault(ctx) if err != nil { log.Fatal(err) } mux := http.NewServeMux() mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte("ok")) }) server := &http.Server{Addr: ":3000", Handler: mux, ReadHeaderTimeout: 5 * time.Second} go func() { if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { log.Fatal(err) } }() <-signals shutdown, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() _ = server.Shutdown(shutdown) _ = client.Close(shutdown) ``` ## Deployment shapes [Section titled “Deployment shapes”](#deployment-shapes) Every production shape runs migrations as a separate deployment step, sets `auto_migrate = false`, and gives API, worker, scheduler, and migration processes independently measured pool sizes. A process runs only the loops it owns. | Shape | HTTP process | Background work | Migration ownership | Important constraint | | ---------------- | ------------------------------------------------ | ------------------------------------------------------------- | ---------------------------------------- | ------------------------------------------------------------------------------------------------------------- | | Container | One foreground server per container | Separate worker/scheduler containers from the same image | Release command or one-shot container | Do not run a process supervisor inside the image | | systemd | `forge-api.service` | `forge-worker.service` and optional `forge-scheduler.service` | Required `forge-migrate.service` oneshot | `TimeoutStopSec` must exceed the Forge drain deadline | | Kubernetes | Deployment with readiness/liveness probes | Separate Deployments, each with its own queue/loop | Pre-deploy Job, awaited before rollout | `terminationGracePeriodSeconds` must exceed drain time | | Serverless HTTP | Module-scoped client reused across warm requests | None inside request handlers | CI/CD job | Use external PostgreSQL/S3; no embedded PostgreSQL, filesystem blobs, workers, scheduler, or maintenance loop | | Dedicated worker | No listener required | One or more named queue workers | Same release migration job | Readiness may be reported to the host through its existing control channel | ### Container commands [Section titled “Container commands”](#container-commands) Build one image with separate entry points. Keep secrets in the platform environment and interpolate them from `forge.toml`. ```dockerfile FROM gcr.io/distroless/cc-debian13 COPY app /app COPY forge.toml /forge.toml USER 65532:65532 ENTRYPOINT ["/app"] CMD ["api"] ``` ```sh /app migrate # release job; exits non-zero unless every target is compatible /app api # HTTP only /app worker emails /app scheduler ``` ### systemd units [Section titled “systemd units”](#systemd-units) ```ini # forge-migrate.service [Service] Type=oneshot User=forge EnvironmentFile=/etc/forge/app.env WorkingDirectory=/srv/forge ExecStart=/srv/forge/app migrate # forge-api.service [Unit] Requires=forge-migrate.service After=network-online.target forge-migrate.service [Service] User=forge EnvironmentFile=/etc/forge/app.env WorkingDirectory=/srv/forge ExecStart=/srv/forge/app api Restart=on-failure TimeoutStopSec=40 ``` Use sibling services for workers and the scheduler. `systemctl stop` sends the process signal; the executable stops dequeuing, drains for 30 seconds, closes Forge, and exits before `TimeoutStopSec`. ### Kubernetes [Section titled “Kubernetes”](#kubernetes) ```yaml apiVersion: batch/v1 kind: Job metadata: { name: forge-migrate-v1-1-0 } spec: template: spec: restartPolicy: Never containers: - name: migrate image: example/app:1.1.0 args: ["migrate"] envFrom: [{ secretRef: { name: forge-runtime } }] --- apiVersion: apps/v1 kind: Deployment metadata: { name: forge-api } spec: replicas: 3 selector: { matchLabels: { app: forge-api } } template: metadata: { labels: { app: forge-api } } spec: terminationGracePeriodSeconds: 40 containers: - name: api image: example/app:1.1.0 args: ["api"] readinessProbe: { httpGet: { path: /readyz, port: 3000 } } livenessProbe: { httpGet: { path: /healthz, port: 3000 } } envFrom: [{ secretRef: { name: forge-runtime } }] ``` Deploy workers and the scheduler as separate Deployments using the same image and different arguments. Scale workers from queue depth and oldest-visible age, not CPU alone. Run one scheduler replica for efficiency or several for availability; PostgreSQL claims and deterministic occurrence IDs make concurrent ticks safe. ## MCP server recipe [Section titled “MCP server recipe”](#mcp-server-recipe) The [Model Context Protocol SDK](https://github.com/modelcontextprotocol/typescript-sdk/blob/main/docs/server.md) owns discovery, schemas, transports, authentication context, cancellation, progress, resources, and protocol errors. Forge supplies ordinary application infrastructure behind tool handlers. This example uses the current TypeScript SDK shape; pin the SDK separately from Forge because its protocol release cadence is independent. ```ts import { McpServer } from "@modelcontextprotocol/server"; import * as z from "zod/v4"; import { ForgeClient } from "forgelib"; const forge = await ForgeClient.init(); const renders = forge.queue<{ invocationId: string; actor: string; inputKey: string; traceparent?: string; audit: { tool: string; requestedAt: string } }>("mcp-render"); const server = new McpServer({ name: "artifact-tools", version: "1.0.0" }); // Wire this into the SDK transport's HTTP authentication hook. The SDK owns the // protocol auth result; Forge validates the application session behind it. async function authenticateMcp(request: Request) { const header = request.headers.get("authorization"); const token = header?.startsWith("Bearer ") ? header.slice(7) : undefined; const actor = token ? await forge.validateSession(token) : null; return actor ? { clientId: actor, scopes: ["tools"], token } : null; } server.registerTool("start_render", { description: "Queue an idempotent artifact render", inputSchema: z.object({ invocationId: z.string().uuid(), inputKey: z.string().min(1) }), annotations: { idempotentHint: true }, }, async ({ invocationId, inputKey }, ctx) => { const actor = ctx.http?.authInfo?.clientId; if (!actor) return { content: [{ type: "text", text: "Authentication required" }], isError: true }; const decision = await forge.rateLimitCheck("mcp-tools", actor, 60, 60, false); if (!decision.allowed) return { content: [{ type: "text", text: "Rate limit exceeded" }], isError: true }; if (ctx.mcpReq.signal.aborted) return { content: [{ type: "text", text: "Cancelled before enqueue" }], isError: true }; const traceparent = ctx.http?.req.headers.get("traceparent") ?? undefined; await renders.enqueue({ invocationId, actor, inputKey, traceparent, audit: { tool: "start_render", requestedAt: new Date().toISOString() } }, { jobId: invocationId, traceContext: traceparent ? { traceparent } : undefined }); await ctx.mcpReq.log("info", { event: "tool_queued", invocationId, actor }); return { content: [{ type: "text", text: `Queued ${invocationId}` }], structuredContent: { invocationId, state: "queued" } }; }); server.registerTool("cancel_render", { description: "Request cancellation of queued or running work", inputSchema: z.object({ invocationId: z.string().uuid() }), annotations: { idempotentHint: true }, }, async ({ invocationId }) => { const status = await renders.cancel(invocationId); return { content: [{ type: "text", text: status ? status.state : "not found" }] }; }); server.registerTool("get_render", { description: "Inspect work and return its artifact when ready", inputSchema: z.object({ invocationId: z.string().uuid() }), annotations: { readOnlyHint: true }, }, async ({ invocationId }) => { const status = await renders.status(invocationId); if (!status || status.state !== "succeeded") return { content: [{ type: "text", text: status?.state ?? "not found" }] }; const key = `mcp-artifacts/${invocationId}`; const download = await forge.blobPresignDownload(key, 300); return { content: [{ type: "resource_link", uri: download.url, name: `${invocationId}.bin`, mimeType: "application/octet-stream" }] }; }); ``` The worker checks Forge’s cooperative cancellation flag between expensive steps, writes the artifact to blob storage, and records immutable audit rows in an application-owned table or transactional outbox. The queue payload carries actor, invocation ID, input reference, and trace correlation, but it is not an audit log. A session token may authenticate the HTTP transport before the SDK builds `authInfo`; Forge does not reinterpret MCP authentication. Long work returns a job handle immediately instead of holding an HTTP request open. If the MCP SDK’s experimental task contract becomes stable, an adapter may map that protocol task to the same queue job without changing Forge’s queue semantics. ## CloudEvents structured JSON [Section titled “CloudEvents structured JSON”](#cloudevents-structured-json) `encodeCloudEvent` / `encode_cloud_event` / `EncodeCloudEvent` and the Rust `CloudEvent::encode` helper implement the CloudEvents 1.0 structured JSON envelope. Binary data always uses standard `data_base64`; decoding accepts both `data_base64` and standard JSON `data`, preserves extension attributes, rejects conflicting data members, and caps the envelope at 1 MiB. Transport bindings, batch format, delivery, retries, and event retention remain outside Forge. * Node ```ts const wire = encodeCloudEvent({ id: job.id, source: "/workers/render", type: "com.example.rendered", datacontenttype: "application/pdf", data: pdf, extensions: { traceid }, }); const event = decodeCloudEvent(wire); ``` * Python ```python wire = forgelib.encode_cloud_event({ "id": job.id, "source": "/workers/render", "type": "com.example.rendered", "datacontenttype": "application/pdf", "data": pdf_bytes, "extensions": {"traceid": trace_id}, }) event = forgelib.decode_cloud_event(wire) ``` * Rust ```rust let mut event = forgelib::CloudEvent::new(job.id.to_string(), "/workers/render", "com.example.rendered"); event.data_content_type = Some("application/pdf".into()); event.data = Some(pdf); let wire = event.encode()?; ``` * Go ```go contentType := "application/pdf" wire, err := forge.EncodeCloudEvent(forge.CloudEvent{ ID: job.ID, Source: "/workers/render", Type: "com.example.rendered", DataContentType: &contentType, Data: pdf, }) ``` The supported profile follows the [CloudEvents JSON event format](https://github.com/cloudevents/spec/blob/main/cloudevents/formats/json-format.md). Use an official CloudEvents SDK when an application needs HTTP binary mode, Kafka bindings, batches, custom attribute types, or envelopes larger than this bounded helper. ## Environment import and export [Section titled “Environment import and export”](#environment-import-and-export) The environment adapter takes explicit maps; it never enumerates or mutates the process environment and never logs values. Each logical config key declares ordered aliases used by common platforms. Import accepts one present value, accepts several aliases only when their values agree, and rejects ambiguous deployments. Export writes the first canonical name. ```ts const mappings = [ { key: "database.url", names: ["DATABASE_URL", "POSTGRES_URL"] }, { key: "blob.bucket", names: ["S3_BUCKET", "AWS_S3_BUCKET"] }, ]; const imported = importEnvConfig(process.env as Record, mappings); for (const [key, value] of Object.entries(imported)) await forge.config(key).set(value); const deploymentEnv = exportEnvConfig(imported, mappings); ``` The same helpers are `import_env_config` / `export_env_config` in Python, `import_env_config` / `export_env_config` in Rust, and `ImportEnvConfig` / `ExportEnvConfig` in Go. Mappings are application configuration because aliases differ between providers and products. Secrets remain platform-owned; export returns a map for a trusted deployment tool and does not write a `.env` file. ## Existing admin and observability surfaces [Section titled “Existing admin and observability surfaces”](#existing-admin-and-observability-surfaces) Forge exposes structured diagnostics and Prometheus text but starts no admin server. Put them behind the application’s existing operator authentication, a private listener, or a service-mesh policy. Do not expose backend messages publicly. ```ts app.get("/internal/forge/diagnostics", requireOperator, async (_req, res) => { const [runtime, scheduler] = await Promise.all([ forge.diagnostics(2), forge.schedulerDiagnostics(), ]); res.json({ runtime, scheduler }); }); app.get("/internal/metrics", requireMetricsPrincipal, (_req, res) => { res.type("text/plain; version=0.0.4").send(forge.renderPrometheus()); }); ``` For an existing OpenTelemetry stack, install the application’s tracing subscriber/provider first. Forge spans flow into it; queue and outbox metadata preserve validated W3C trace context. Keep raw keys, principals, tool arguments, prompts, responses, and signed URLs out of metric labels and span attributes. Correlate them with bounded opaque IDs in application logs or audit storage instead. Diagnostics are operator data `diagnostics` performs bounded live backend checks. Give it a short deadline, rate-limit it, and authenticate it independently of public health checks. `/healthz` should remain a cheap local liveness response; readiness may probe required backends with its own deadline. # Operations > Running Forge in production: the system database, workers, the scheduler, health checks, durability, and error handling. Forge runs inside your process. It doesn’t start threads or daemons on its own, so the background work, the scheduler, and the maintenance sweep are loops you run. This page covers what to run and what to watch once Forge is behind real traffic. ## Deploy migrations before the application [Section titled “Deploy migrations before the application”](#deploy-migrations-before-the-application) Forge owns the objects listed in [`contract/schema-ownership.json`](https://github.com/isala404/forge/blob/main/contract/schema-ownership.json). Point `[postgres]` at a database or schema reserved for Forge and do not write its objects directly. Development and test environments migrate automatically by default. Production defaults to `auto_migrate = false`, so application startup validates the schema and fails before serving traffic when migrations are pending, locked, incompatible, or drifted. Run the migration API as a deployment job first: * Node ```ts const reports = await ForgeClient.migrate(); for (const report of reports) console.error(report.target, report.state, report.message); if (reports.some((report) => report.state !== "applied")) process.exit(1); ``` * Python ```python reports = await forgelib.ForgeClient.migrate() for report in reports: print(report.target, report.state, report.message) if any(report.state != "applied" for report in reports): raise SystemExit(1) ``` * Rust ```rust let reports = Forge::migrate().await?; for report in &reports { eprintln!("{}: {} ({})", report.target, report.state.as_str(), report.message); } anyhow::ensure!(reports.iter().all(|report| report.is_compatible()), "migration failed"); ``` * Go ```go reports, err := forge.Migrate(ctx) if err != nil { return err } for _, report := range reports { log.Printf("%s: %s (%s)", report.Target, report.State, report.Message) if report.State != "applied" { os.Exit(1) } } ``` `migrationStatus`/`migration_status`/`MigrationStatus` is read-only and also reports the safe PostgreSQL session identity when another migrator owns the advisory lock. `validateSchema`/`validate_schema`/`ValidateSchema` checks migration history without taking the lock. Every distinct measured database target receives one report. `migration_lock_timeout_secs` bounds lock acquisition and defaults to 30 seconds. Forge starts from one `v001_schema` baseline. Unknown migration history and checksum drift are incompatible. Future schema changes append a migration; existing migration files remain immutable after release. ## Workers [Section titled “Workers”](#workers) A queue holds jobs until a worker leases them. Run a worker per queue. Each binding has a managed worker that handles leasing, heartbeats, acks, retries, and graceful shutdown. * Node ```ts type SendEmail = { to: string }; await forge.worker("emails", async (job) => { await send(job.payload); }, { concurrency: 8, visibilitySeconds: 60, heartbeatSeconds: 15, retryBackoffSeconds: 0.25, drainDeadlineSeconds: 30, identity: "email-worker", signal: shutdown.signal, onError: report }); ``` * Python ```python # leases, heartbeats, acks on return, nacks on exception await forge.worker( "emails", handle, concurrency=8, visibility_seconds=60, heartbeat_seconds=15, retry_backoff_seconds=0.25, drain_deadline_seconds=30, identity="email-worker", stop=shutdown, on_error=report, ) ``` * Rust ```rust // bounded concurrency, auto-heartbeat, ack on Ok / nack on Err, drains on shutdown forge.worker("emails") .concurrency(8) .visibility_timeout(Duration::from_secs(60)) .heartbeat_cadence(Duration::from_secs(15)) .retry_backoff(Duration::from_millis(250)) .grace(Duration::from_secs(30)) .identity("email-worker") .on_error(report) .run(|job| async move { send(&job).await }) // returns Result<(), E> .await; ``` * Go ```go err := forgeClient.RunWorker(ctx, "emails", handle, forge.WorkerOptions{ Concurrency: 8, Visibility: 60 * time.Second, HeartbeatCadence: 15 * time.Second, RetryBackoff: 250 * time.Millisecond, DrainDeadline: 30 * time.Second, Identity: "email-worker", OnError: report, }) ``` The shared worker state machine is `polling → handling/heartbeating → settling → draining → stopped`. Cancellation stops polling first. If cancellation wins the same boundary as dequeue, the fresh lease is released without starting user code. Active handlers drain until the configured deadline, then receive the language-native cancellation mechanism and their leases are released. Lease loss cancels the stale handler and prevents it from acknowledging or extending the newer owner’s lease. Backend retry waits use bounded 80–120% jitter. Worker identity appears only in diagnostics and never as a default metric label. Run as many worker processes as you need; jobs are leased, so two workers never get the same job at once. A handler that crashes lets the lease expire and the job comes back, which is why handlers need to be safe to run twice. Jobs that exhaust their attempts move to `".dlq"`. Queue depth reports visible, in-flight, delayed, and oldest-visible age. Default metrics cover dead-letter transitions, redrives, lease loss, handler outcomes, queue depth/age, and outbox backlog without job IDs, payloads, worker identities, or queue names as dimensions. Operator controls are typed library calls, not a second admin server. Batch enqueue returns per-item results; batch dequeue leases at most ten jobs with independent receipts. Pause/resume retains queued work and affects only new claims. Inspection, status filtering, counter-based throughput estimates, dead-letter redrive, and confirmed purge can be exposed through the application’s existing admin authorization boundary. ## Transactional outbox [Section titled “Transactional outbox”](#transactional-outbox) Use the versioned [`sql/forge_outbox_v1.sql`](https://github.com/isala404/forge/blob/main/sql/forge_outbox_v1.sql) contract when a domain change and queue enqueue must be atomic. Copy that application-owned table into the application’s migration, then insert its row with the application’s own PostgreSQL driver in the same transaction as the domain write. Forge does not migrate or own this table; grant the Forge runtime role `SELECT` and `UPDATE` on it. ```sql BEGIN; UPDATE orders SET state = 'paid' WHERE id = $1; INSERT INTO app_forge_outbox_v1 (event_id, namespace, destination, payload) VALUES ($2, 'billing', 'receipts', convert_to($3, 'UTF8')); COMMIT; ``` Run one relay per replica with `runOutboxRelay` / `run_outbox_relay` / `run_outbox_relay_until` / `RunOutboxRelay`, passing the host language’s cancellation input. Claims use a bounded lease and `FOR UPDATE SKIP LOCKED`. The event UUID becomes the queue job UUID. A crash before enqueue leaves a reclaimable outbox row; a crash after enqueue but before marking the row dispatched repeats the deterministic enqueue and returns the same job ID; a crash after marking is complete. This is at-least-once relay processing with one effective queue row, not exactly-once handler execution. Invalid rows are retried with a bounded safe summary; payloads and trace baggage are never logged. ## The scheduler [Section titled “The scheduler”](#the-scheduler) Cron and one-off schedules only fire when something ticks the scheduler. The example apps run one loop that ticks the scheduler and then runs the maintenance sweep, on every replica. Each due job is claimed once no matter how many replicas tick at the same time. * Node ```ts setInterval(async () => { try { await forge.runSchedulerOnce(); await forge.maintain(); } catch (e) { console.warn("scheduler tick failed:", e); } }, 30_000); ``` * Python ```python async def scheduler(): while True: try: await forge.run_scheduler_once() await forge.maintain() except Exception as e: log.warning("scheduler tick failed: %s", e) await asyncio.sleep(30) ``` * Rust ```rust use std::time::Duration; loop { if let Err(e) = forge.run_scheduler_once().await { tracing::warn!(error = %e, "scheduler tick failed"); } if let Err(e) = forge.maintain().await { tracing::warn!(error = %e, "maintenance failed"); } tokio::time::sleep(Duration::from_secs(30)).await; } ``` * Go ```go ticker := time.NewTicker(30 * time.Second) defer ticker.Stop() for { if _, err := forge.RunSchedulerOnce(ctx, 1000); err != nil { log.Printf("scheduler tick failed: %v", err) } if err := forge.Maintain(ctx); err != nil { log.Printf("maintenance failed: %v", err) } select { case <-ctx.Done(): return; case <-ticker.C: } } ``` Use `schedulerDiagnostics` / `scheduler_diagnostics` / `Schedule::diagnostics` / `SchedulerDiagnostics` for alerts. `lag` is the age of the oldest unpaused due row, `dueCount` is the current backlog, `lastSuccessfulTick` advances only after a complete pass, and `enqueueFailures` is cumulative. Alert on sustained lag or a non-advancing successful tick rather than on one transient failure. Pause and resume operate on one named schedule and preserve its payload and policy for inspection. `maintain()` is the housekeeping sweep: it purges expired key/value rows and old finished jobs, reclaims leases left behind by crashed workers, drops stale rate-limit and dedup rows, and expires dead sessions and one-time tokens. It’s idempotent, so running it on every replica is fine. Rust also has `run_scheduler()`, a managed loop that ticks every 30 seconds on its own if you’d rather run the sweep separately. ## Durability [Section titled “Durability”](#durability) The Postgres backend is durable and shared. The memory backend is neither, which is the point for tests but a trap in production. Pub/sub and rate limiting in particular depend on every replica seeing the same state, so on `memory` they only work within one process. Forge logs a warning when either resolves to memory, since that’s almost always a misconfiguration outside of tests. Keep production on `postgres` unless you have injected your own backend. ## Health checks [Section titled “Health checks”](#health-checks) Liveness and readiness are separate. `isLive()` / `is_live()` / `is_live` / `IsLive()` only reports whether the local handle is open. `probe()` / `Probe()` runs one real operation against every enabled backend under one total deadline and returns redacted status, latency, error category, and last-success time per backend. By default every backend decides readiness; pass an explicit readiness set when a dependency is intentionally optional. `diagnostics()` / `Diagnostics()` adds configuration, PostgreSQL version, migration state, runtime permissions, backend reachability, application/database clock skew, and unsafe-production-setting checks. It returns stable pass/warn/fail records with redacted messages. Applications expose or record this report through their existing deployment job, test suite, observability system, or authenticated admin surface; Forge does not create an unauthenticated diagnostics endpoint. `backendCapabilities()` lists which provider backs each primitive, whether it is durable, and any caveats. It is synchronous configuration inspection, not a health check. * Node ```ts const report = forge.backendCapabilities(); // [{ primitive, provider, durable, caveats }] if (!forge.isLive()) return new Response("stopping", { status: 503 }); const health = await forge.probe(2); ``` * Python ```python report = forge.backend_capabilities() # synchronous; do not await # [BackendInfo(primitive, provider, durable, caveats), ...] if not forge.is_live(): return Response(status_code=503) health = await forge.probe(deadline_seconds=2.0) ``` * Rust ```rust let report = forge.backend_capabilities(); // each entry: primitive, provider, durable, caveats if !forge.is_live() { /* return 503 */ } let health = forge.probe(ProbeOptions::default()).await?; ``` * Go ```go report := forge.BackendCapabilities() if !forge.IsLive() { /* return 503 */ } health, err := forge.Probe(ctx, forgelib.ProbeOptions{Deadline: 2 * time.Second}) ``` ## Metrics and tracing [Section titled “Metrics and tracing”](#metrics-and-tracing) Metrics belong to each Forge instance. `metricsSnapshot()` / `metrics_snapshot()` / `metrics_snapshot` / `MetricsSnapshot()` returns bounded samples, while `renderPrometheus()` / `render_prometheus()` / `render_prometheus` / `RenderPrometheus()` emits Prometheus text without installing a collector or starting an exporter. Operations, pool use, retries, queue and worker state, scheduler and maintenance work, config sources, blob transfers, authentication, rate-limit decisions, and outbox backlog use fixed low-cardinality labels. Raw keys, user IDs, paths, queue names, job IDs, topics, model names, error messages, bodies, prompts, model output, content, tokens, and signed URLs are never labels or span attributes. Queue and outbox propagation stores validated W3C `traceparent` and `tracestate` beside the payload. Baggage is empty by default and only explicit allow-listed keys cross the boundary. Forge creates producer, receive/process, settle, scheduler, relay, and blob spans with fixed semantic names when tracing is enabled. Applications remain responsible for model-call spans and may attach GenAI attributes through their OpenTelemetry instrumentation; Forge never inspects model payloads. ## Errors and retries [Section titled “Errors and retries”](#errors-and-retries) Forge failures carry the same code, retryable flag, safe message, operation, and optional backend identifier in Rust, JavaScript, Python, and Go. Only `Unavailable` and a `Backend` error explicitly marked retryable may be retried; worker and reconnect loops use that same rule. `Config`, `NotConfigured`, `NotFound`, `Invalid`, `Limit`, and `Precondition` require a changed request or refreshed state. The displayed message is safe and excludes raw database/provider errors, while the local cause chain remains available for debugging. The full table is in the [Reference](/reference/#errors). Note A rate-limit denial and an expired session are not errors. `rateLimitCheck` returns a decision with `allowed: false`, and `validateSession` returns nothing for a bad token. Check the return value rather than catching an exception. ## Scaling a primitive out [Section titled “Scaling a primitive out”](#scaling-a-primitive-out) When one primitive outgrows the shared database, give it its own. Point it at a separate Postgres in `forge.toml` with a `[databases.]` table and it runs on an isolated pool, on its own server if you want, with no change to your code. See [Configuration](/configuration/#a-separate-database-per-feature). ## Known limitations [Section titled “Known limitations”](#known-limitations) These limits matter in production. **Queue receipts are process-local in Node and Python.** A receipt is only meaningful inside the process that dequeued the job: the binding holds the lease in an in-process map, so `queueAck`, `queueNack`, and `queueHeartbeat` must run in that same process. You can’t hand a receipt to another worker or persist it and settle the job later. A job left unsettled is reclaimed the usual way, by lease expiry on the server, and redelivered. Rust and Go pass the backend lease receipt directly and do not have this binding-map restriction. **Rate limiting on `memory` counts per process.** Each process keeps its own buckets, so across replicas the limit is effectively multiplied by the replica count and every check under-counts. Use `postgres` for any limit that has to hold across processes. **Pub/sub is at-most-once and non-durable on every backend.** A message reaches only the subscribers connected at publish time; there’s no persistence, replay, or delivery guarantee. On `memory` it’s in-process only, so nothing crosses replicas. Treat it as a notification, not a log, and reconcile durable state through the database. **Filesystem blob splits bytes from metadata.** Objects live as files under the configured root while their metadata rows live in Postgres, so a `put` is two writes and is not atomic with your application’s SQL. A crash between them leaves a metadata row pointing at a missing file until the maintenance sweep reclaims it. Multiple replicas need a shared mount, since one replica can’t read files another wrote to local disk. **S3 policy remains provider-owned.** Forge sends object bytes, web metadata, conditional request headers, provider-managed encryption headers, and multipart commands. Configure lifecycle expiration, retention, replication, versioning, object lock and legal hold, CDN caching and invalidation, CORS, public-access blocks, bucket policy, IAM, and KMS grants in the provider. Roll out those policies independently, verify them with provider tooling, and do not infer them from Forge `head` results. **Node `kvIncr` returns a JavaScript number.** The counter is an `i64` in the core, but JS numbers are `f64`, so a value past 2^53 loses precision on the way back. Real counters never reach that range; if yours might, read it back losslessly with `kvGetBytes`. The Python binding returns the exact integer. # Performance and scaling > Repeatable Forge benchmarks, regression budgets, contention tests, pool sizing, and evidence gates for larger architecture. Forge performance work starts with a report, not a topology. The repository owns a release-mode harness, JSON budget file, and checker. Budgets are regression ceilings on comparable hardware, not capacity promises for an application. ## Run the backend harness [Section titled “Run the backend harness”](#run-the-backend-harness) The `smoke` profile uses 20 iterations, four concurrent tasks, and two PostgreSQL child processes. The `full` profile uses 500 iterations, 16 tasks, 1 MiB blobs, and four child processes. Both measure p95 queue round trips, queue throughput, same-subject rate-limit contention, blob put/get/delete transfer, maintenance sweeps, and allocator activity. PostgreSQL reports peak pool use and starts separate OS processes against one temporary database; this catches connection, row-lock, lease, and rate-limit contention that an async microbenchmark cannot. ```bash cargo bench --bench performance -- --backend memory --profile full --output memory.json python3 tools/performance/check.py memory.json ``` For an installed PostgreSQL server, `PERF_DATABASE_URL` is a maintenance database whose user may create and drop databases. The harness creates a UUID-named database, migrates it, gives every child the same credential through its environment rather than command-line arguments, closes every pool, and force-drops only that generated database. ```bash PERF_DATABASE_URL='postgres://postgres:postgres@127.0.0.1:5432/postgres' \ cargo bench --bench performance -- --backend postgres --profile full --output postgres.json python3 tools/performance/check.py postgres.json ``` Rust users can instead exercise Forge’s embedded PostgreSQL build. `FORGE_PERF_EMBEDDED_DIR` defaults to `target/forge-perf-postgres`. ```bash cargo bench --features embedded --bench performance -- \ --backend embedded --profile full --output postgres.json cargo bench --features embedded --bench performance -- \ --backend embedded-filesystem --profile full --output filesystem.json cargo bench --features embedded --bench performance -- \ --backend embedded-s3 --profile full --output s3.json ``` Filesystem blob measurements use the same temporary PostgreSQL database for Forge metadata and default to `target/forge-perf-blobs`. S3 measurements require a disposable bucket or prefix and use the same workload. The harness deletes every measured object; provider lifecycle policy remains a separate operator concern. ```bash PERF_DATABASE_URL="$DATABASE_URL" cargo bench --bench performance -- \ --backend filesystem --profile full --output filesystem.json PERF_DATABASE_URL="$DATABASE_URL" PERF_S3_ENDPOINT=http://127.0.0.1:9000 \ PERF_S3_BUCKET=forge-test PERF_S3_ACCESS_KEY=forge PERF_S3_SECRET_KEY=secret \ cargo bench --bench performance -- --backend s3 --profile full --output s3.json ``` The canonical ceilings live in `benchmarks/budgets.json`. Validate its coverage without running a workload with `python3 tools/performance/check.py --validate-only`. A benchmark regression must be reproduced on the same runner before changing a budget. ## Measure language boundaries [Section titled “Measure language boundaries”](#measure-language-boundaries) The boundary workload encodes and decodes the same small CloudEvents 1.0 value. It isolates package/runtime overhead from database latency and emits the same checked JSON report in every supported runtime. Run release or installed package artifacts rather than source-only substitutes. ```bash cargo run --release --example performance_boundary -- --output rust.json node tools/performance/boundary.mjs --output node.json bun tools/performance/boundary.mjs --output bun.json python tools/performance/boundary.py --output python.json go -C bindings/go run ./cmd/perf-boundary --output go.json python3 tools/performance/check.py rust.json ``` Node/Bun require the binding’s normal `npm run build`; Python requires the wheel or editable package installed into the selected interpreter. Go is native. The budget file names Rust, Node, Bun, Python, and Go separately so a fast core cannot hide a boundary regression. ## Read the numbers correctly [Section titled “Read the numbers correctly”](#read-the-numbers-correctly) * Queue latency is enqueue, immediate lease, and acknowledgement. Throughput counts completed round trips, not bare inserts. * Rate-limit contention uses one bucket and subject. This intentionally serializes the hot row and exposes the worst common contention shape. * Blob throughput counts bytes written and read. It includes provider and network behavior for S3. * Allocation counts come from an instrumented system allocator in the benchmark executable. Forge production binaries do not install it. * Maintenance measures one complete idempotent sweep across the configured primitives. * PostgreSQL pool use samples checked-out connections every millisecond. Multi-process latency is elapsed wall time per worker iteration, with every process performing a rate-limit check and a complete queue round trip. Do not compare debug to release, smoke to full, local S3 to a hosted region, or different CPUs as if they were regressions. Preserve the JSON report with the runner, Forge revision, PostgreSQL/provider version, and hardware description when making a scaling decision. ## Pool sizing [Section titled “Pool sizing”](#pool-sizing) PostgreSQL allocates resources from `max_connections`, so increasing it is not free. Keep an operator reserve of at least five connections or 10%, whichever is larger, and budget every Forge process explicitly. See PostgreSQL’s [connection settings](https://www.postgresql.org/docs/current/runtime-config-connection.html). Use this deployment calculation: ```text required server slots = operator reserve + sum(process replicas × configured pool max) + subscribed process replicas + migration job pool max while it runs ``` Each PostgreSQL Forge handle uses one pool slot for config invalidation. The first pub/sub subscription opens one additional dedicated server connection outside the pool, shared across that handle’s topics. Start small, measure `forge_pool_connections{state="open|idle"}`, acquisition timeouts, database saturation, and handler concurrency, then raise one role at a time. | Process role | Starting pool | Reason | | ------------- | -------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------- | | API | 4 | One config-listener slot plus three concurrent database operations; cap request concurrency separately. | | Worker | `min(worker concurrency, 8) + 1` | Handlers do not hold a connection for their whole runtime; the extra slot serves config invalidation. Measure actual overlapping Forge calls. | | Scheduler | 2 | One tick transaction plus config invalidation. Multiple replicas coordinate in PostgreSQL and do not need large pools. | | Listener | 2, plus one server slot | One pooled config listener, one publish/query slot, and the dedicated pub/sub connection outside the pool. | | Migration job | 2 | Forge requires two connections when startup migration is enabled. Run one bounded deployment job, then release both. | If a process owns several roles, use one Forge handle and size for overlapping work, not the sum of independent guesses. Separate databases or pools are justified only after the pool report and PostgreSQL wait events identify a hot primitive. ## Architecture gates [Section titled “Architecture gates”](#architecture-gates) Forge keeps explicit batch enqueue/dequeue APIs, but it does not adapt batch size behind the caller. The measured smoke workload shows millisecond PostgreSQL round trips without evidence that hidden batching would improve the complete enqueue/lease/settle path. Revisit only when the full profile shows a sustained throughput miss and an experiment improves throughput by at least 25% while p95 latency stays within budget, per-item errors remain visible, and crash recovery is unchanged. Forge ships no partitioning migration. First record real row counts, `pg_total_relation_size`, retention-delete cost, autovacuum behavior, and query plans. Consider partitioning only when a Forge table reaches roughly 10 million live rows or 10 GiB and measurements show retention or index maintenance is the bottleneck. The partition key must align with deletion and every hot query before an expand/contract migration is proposed; table size alone is not proof. Forge routes no operations to read replicas. PostgreSQL hot standbys can lag and can cancel conflicting queries, so stale routing is a public consistency decision, not a pool flag. See PostgreSQL’s [hot standby behavior](https://www.postgresql.org/docs/current/hot-standby.html). Only explicitly approximate queue depth/stats and explicitly stale config snapshots are candidates. KV reads, auth, config evaluation, rate limits, queue lease/settlement/status, schedules, migrations, and read-after-write blob metadata stay on the primary. Add replica configuration only after a read-heavy profile proves those candidate operations dominate and application tests accept bounded staleness. # Primitives > The eight Forge primitives, what each one is for, and the methods you'll actually call in Rust, JavaScript, Python, and Go. Forge gives you eight primitives on one client. Each one covers the part of a familiar service that most apps actually use, with the same behavior across Rust, JavaScript, Python, and Go. This page is the tour. Every method, return type, and limit is in the [Reference](/reference/). A few things hold for all of them. Keys and names are byte-limited and oversized input is an error rather than a silent truncation. Every primitive can be prefixed with a [namespace](/configuration/#namespaces) so several apps share one database. Failures map to a small [error taxonomy](/reference/#errors) that tells you whether a retry might help. ## Key-value [Section titled “Key-value”](#key-value) Small values, counters, and cached state. Strings up to 1 MiB with optional TTLs, atomic counters, compare-and-swap, and prefix scans. Lineage: Redis. * Node ```ts // set with a one-hour TTL await forge.kvSet("session:abc", token, 3600); const value = await forge.kvGet("session:abc"); // string | null const n = await forge.kvIncr("views:42", 1); // atomic // write only if absent: a one-line uniqueness guard const claimed = await forge.kvSet("slug:abc", body, null, true); // false if taken ``` * Python ```python # set with a one-hour TTL await forge.kv_set("session:abc", token, 3600) value = await forge.kv_get("session:abc") # str | None n = await forge.kv_incr("views:42", 1) # atomic # write only if absent: a one-line uniqueness guard claimed = await forge.kv_set("slug:abc", body, None, True) # False if taken ``` * Rust ```rust use std::time::Duration; use forgelib::{SetOpts, SetMode}; // set with a one-hour TTL forge.kv().set("session:abc", token.into(), SetOpts::new().with_ttl(Duration::from_secs(3600))).await?; let value = forge.kv().get("session:abc").await?; // Option let n = forge.kv().incr("views:42", 1).await?; // atomic, returns i64 // write only if absent: a one-line uniqueness guard let claimed = forge.kv() .set("slug:abc", body.into(), SetOpts::new().with_mode(SetMode::IfNotExists)) .await?; // false if already taken ``` Keys go up to 512 bytes, values up to 1 MiB. It’s a string and counter store, not a blob store. Expired keys read as missing. `incr` on a non-numeric value is an error. For binary values that aren’t valid UTF-8, use the bytes variants (`kvSetBytes` / `kv_set_bytes` and their getters). ## Queue [Section titled “Queue”](#queue) Durable background jobs. At-least-once delivery, visibility-timeout leasing, retries with a dead-letter queue, delays, and dedup. Lineage: AWS SQS. * Node ```ts const emails = forge.queue("emails"); // producer const effectiveId = await emails.enqueue(msg, { maxAttempts: 5, jobId: eventId, // optional UUID: retry-safe producer identity dedupId: contentHash, // optional short-window content collapse }); // consumer: lease a job, do the work, settle it by receipt const job = await emails.dequeue({ visibilitySeconds: 30, waitSeconds: 20 }); if (job) { try { await send(job.payload); await job.ack(); } catch { await job.nack({ failureSummary: "email provider rejected request" }); } } ``` * Python ```python emails = forge.queue("emails") # producer effective_id = await emails.enqueue( msg, max_attempts=5, job_id=event_id, dedup_id=content_hash ) # consumer: lease a job, do the work, settle it by receipt job = await emails.dequeue(visibility_seconds=30, wait_seconds=20) if job: try: await send(job.payload) await emails.ack(job.receipt) except Exception: await emails.nack(job.receipt) ``` * Rust ```rust use forgelib::{EnqueueOpts, DequeueOpts, NackOpts}; // producer forge.queue().enqueue("emails", body.into(), EnqueueOpts::new().with_max_attempts(5)).await?; // consumer: lease a job, do the work, settle it if let Some(job) = forge.queue().dequeue("emails", DequeueOpts::new()).await? { match send(&job).await { Ok(()) => forge.queue().ack(&job).await?, Err(_) => forge.queue().nack(&job, NackOpts::default()).await?, } } ``` * Go ```go effectiveID, err := forgeClient.Enqueue(ctx, "emails", body, forge.EnqueueOptions{ ID: eventID, DedupID: contentHash, MaxAttempts: 5, }) job, err := forgeClient.Dequeue(ctx, "emails", forge.DequeueOptions{ Visibility: 30 * time.Second, Wait: 20 * time.Second, }) if err == nil && job != nil { if err := send(job.Payload); err == nil { err = forgeClient.Ack(ctx, job.Receipt) } else { err = forgeClient.Nack(ctx, job.Receipt, forge.NackOptions{FailureSummary: "email provider rejected request"}) } } ``` At-least-once, so handlers must be idempotent A job can be delivered more than once (a worker that crashes mid-job lets the lease expire and the job comes back). Write handlers that are safe to run twice. A job that fails its `max_attempts` (5 by default) lands on `".dlq"`. In Node and Python you settle a job by its `receipt`, which is unique per delivery; the `id` is stable across redeliveries and makes a good idempotency key. Payloads are raw bytes up to 256 KiB; JSON codecs are binding conveniences, not a storage requirement. The optional version-1 queue-envelope helpers validate the complete encoded size plus bounded schema, content type, correlation ID, trace context, and artifact-reference metadata before enqueue. Store larger request bodies or generated artifacts in the blob primitive and enqueue a stable blob reference. CloudEvents conversion stays application-owned because Forge does not use CloudEvents as its internal format, and Forge intentionally defines no prompt, completion, tool-call, model-provider, workflow-graph, fan-in, or compensation DTOs. A handler that might outlive its lease should `heartbeat` to extend it. Cancellation is cooperative: queued or delayed work becomes cancelled atomically, while a leased job becomes cancel-requested and the managed worker propagates that through Rust’s cancellation token, JavaScript’s `AbortSignal`, Python task cancellation plus an event, or Go’s `Context`. Forge cannot terminate application code, so handlers must observe that signal. Cancellation and redelivery fence the old receipt; a late ack or heartbeat fails with `Precondition`. Status lookup exposes only queued, delayed, leased, retrying, succeeded, dead, cancel-requested, and cancelled. Operator listing is bounded and filterable. `queue.payload_retention_secs` defaults to one day and clears bytes from terminal jobs. `queue.succeeded_retention_secs`, `queue.dead_retention_secs`, and `queue.cancelled_retention_secs` default to seven days and independently remove terminal status. `queue.terminal_retention_secs` sets all three before an individual override. Batch enqueue returns one ordered result per item and does not roll successful siblings back when another item fails. Retry only failed items with deterministic IDs. Batch dequeue returns at most ten independently fenced leases and long-polls only for the first claim. Pause prevents new leases after the operator write commits; it preserves queued work and does not revoke existing leases. Queue stats combine monotonic counters with an indexed oldest-visible lookup, so throughput estimates do not scan retained payload rows. Priority is deliberately bounded to low, normal, or high. A backend preserves FIFO within one priority; continuously arriving higher-priority work can starve lower-priority work, so reserve high priority for bounded urgent traffic. An optional concurrency key and per-key dequeue limit prevent one tenant, resource, or provider from occupying every worker slot; a skipped noisy key does not block other eligible keys. `dedup_id` collapses repeated enqueues within a window. A caller-selected job UUID is the primary idempotency key: repeating it on the same namespace and queue returns that effective ID without another insert; using it on another queue is `Precondition`. If a live dedup reservation points at a different requested job ID, enqueue is also `Precondition`. Permanent failure or cancellation releases the dedup reservation so the content can be intentionally submitted again. Dead-letter inspection returns payload-free operator metadata: job ID, source queue, failed-attempt count, enqueue/dead-letter timestamps, and a safe error summary capped at 512 bytes. Redrive requires both an explicit destination and an explicit `clear` or `preserve` dedup policy. Purge is split into a dry-run count and a destructive call whose confirmation must exactly equal the source queue. A dead letter can be redriven only while its payload is retained; its payload-free status remains inspectable until terminal retention removes the row. Each binding also ships a managed worker (`forge.worker("emails", handler)` in Node/Python, `forge.worker("emails")` in Rust, and `RunWorker` in Go) covered in [Operations](/operations/#workers). ## Pub/sub [Section titled “Pub/sub”](#pubsub) Live fan-out to whoever is connected: presence, typing indicators, a counter that updates on a dashboard. It is not durable. Delivery is at-most-once and only reaches current subscribers. Lineage: Postgres LISTEN/NOTIFY and Redis pub/sub. When a message must not be lost, use the queue. * Node ```ts const room = forge.topic("room:42"); await room.publish(msg); for await (const event of await room.subscribe()) { // forward payload to the connected client } ``` * Python ```python room = forge.topic("room:42") await room.publish(msg) async for event in room.subscribe(): ... # forward payload to the connected client ``` * Rust ```rust use futures_util::StreamExt; forge.pubsub().publish("room:42", msg.into()).await?; let mut sub = forge.pubsub().subscribe("room:42").await?; while let Some(Ok(payload)) = sub.next().await { // forward payload to the connected client } ``` Payloads are UTF-8 and capped near 7 KB, the Postgres NOTIFY limit. For anything larger, publish a reference such as a row id and let the subscriber read it. A message sent while a subscriber is offline is not kept for it. ## Blob [Section titled “Blob”](#blob) Object storage for uploads and generated files: put, get, delete, copy, list, ranged and streaming access, conditional reads and writes, checksums, S3 multipart handles, and proxy or native presigned requests. Lineage: S3. Objects live in PostgreSQL by default, on the local filesystem for local development, or in S3-compatible storage for production-scale bytes. * Node ```ts await forge.blobPut("avatars/42.png", png, "image/png"); const data = await forge.blobGet("avatars/42.png"); // string | null // for binary, blobGetBytes returns a Buffer const url = (await forge.blobPresignDownload("avatars/42.png", 3600)).url; ``` * Python ```python await forge.blob_put("avatars/42.png", png, "image/png") # png is bytes data = await forge.blob_get("avatars/42.png") # bytes | None url = (await forge.blob_presign_download("avatars/42.png", 3600)).url ``` * Rust ```rust use std::time::Duration; use forgelib::PutOpts; forge.blob().put("avatars/42.png", bytes, PutOpts::new().with_content_type("image/png")).await?; let data = forge.blob().get("avatars/42.png").await?; // Option // a link the browser can use directly, good for an hour let url = forge.blob() .presign_download("avatars/42.png", Duration::from_secs(3600)).await?.url; ``` * Go ```go checksum := sha256.Sum256(png) expected := hex.EncodeToString(checksum[:]) err := forgeClient.BlobPut(ctx, "avatars/42.png", png, forge.PutOptions{ ContentType: "image/png", CacheControl: "public, max-age=300", ContentDisposition: "inline", ChecksumSHA256: expected, }) if err != nil { return err } info, err := forgeClient.BlobHead(ctx, "avatars/42.png") if err != nil { return err } result, err := forgeClient.BlobGetIf(ctx, "avatars/42.png", nil, &info.ETag) ``` The buffered `put`, `get`, and ranged-read surfaces are capped at 50 MiB. Use `put_stream`/`blobPutFile`/`blob_put_file`/`BlobPutStream` for larger uploads and read large objects in bounded ranges; Rust and Go also expose a streaming reader. `list` returns `BlobSummary` values in lexicographic key order, while `head` returns full `BlobInfo` including content type, `Cache-Control`, `Content-Disposition`, a Forge SHA-256 checksum when recorded, server-side encryption mode when reported, and user metadata. Cursors are opaque provider tokens, listing is not a point-in-time snapshot, and concurrent writes can appear or disappear between pages. ETags are opaque version tokens, not guaranteed MD5 digests. Conditional `get_if` reads the body and checks `if_match` or `if_none_match` in one backend operation. A match failure is `PRECONDITION`; an unchanged `if_none_match` read returns `not_modified`; an absent key returns `missing`. Put accepts `CreateOnly` or `MatchVersion(etag)` and never emulates a conditional write with a prior read. The portable `copy` helper streams a source into a destination and preserves content type, metadata, cache control, and disposition unless replaced. It is deliberately non-atomic: the source can change while it is copied. Forge has no `move` helper because a portable copy followed by an unconditional delete cannot honestly provide atomic move or protect a newer source version. Forge uses one portable checksum algorithm: SHA-256 encoded as 64 lowercase hexadecimal characters. Buffered put can require an expected checksum and records the computed checksum independently of ETag. `verify_checksum_sha256` streams the authoritative bytes and returns whether they match, including after an upload whose provider did not retain Forge checksum metadata. Forge does not advertise a checksum selector because every supported backend must verify every advertised algorithm. Provider-native server-mediated multipart handles are available only with the S3 backend. Create an upload, send numbered parts, retry a failed part by uploading the same number again, complete with 1–10,000 strictly ordered receipts, or abort the handle. Completion applies the destination precondition; abort is idempotent once the provider reports the upload absent. PostgreSQL, filesystem, and memory return `NOT_CONFIGURED` for these handles because their streaming put already accepts an unknown-length reader without a resumable provider session. A multipart expected checksum is not accepted up front; call `verify_checksum_sha256` after completion. S3 puts, multipart creation, and native PUT presigns can request provider-managed `AES256` encryption or `aws:kms` with a provider key ID. Forge forwards the encryption headers and reports the mode from `head`; it never accepts, stores, rotates, or grants encryption key material. Whether a provider honors a KMS key, and who may decrypt it, remains bucket and IAM policy. Forge proxy presigns require `signing_secret` and return structured `ProxyPresign` data. The proxy signature covers its format version, namespace, method, logical key, expiry, and upload-size ceiling. S3 native GET and PUT presigns are separate `NativePresign` values with required headers and constraints; native PUT cannot portably enforce a maximum body size. Both URL forms are bearer credentials: never log their query strings. Object lifecycle rules, retention and deletion protection, replication, legal hold, CDN behavior, public access, CORS, bucket policy, IAM, and KMS key administration are provider-owned. Configure and audit them in the object-storage provider; Forge does not mirror them into its database or claim that deleting Forge metadata changes those policies. ## Auth [Section titled “Auth”](#auth) Passwords, sessions, API keys, and one-time tokens. argon2id hashing, opaque session tokens with sliding idle and hard absolute deadlines, `fk_` API keys, and single-use purpose-scoped tokens for password reset, email verification, and magic links. Lineage: OWASP, the PHC string format, and the Stripe/GitHub key style. Forge stores only hashes and never owns your users table. * Node ```ts const hash = await forge.hashPassword(password); // store this const ok = await forge.verifyPassword(password, hash); const token = await forge.createSession(userId); const userId2 = await forge.validateSession(token); // string | null // one-time token: email the link, then consume it exactly once const reset = await forge.createToken(userId, "password-reset", 900, Buffer.from("return=/settings")); const consumed = await forge.consumeToken(reset, "password-reset"); // { userId, payload } | null const apiKey = await forge.createApiKeyWith(userId, "automation", 30 * 86400, ["deploy"], { environment: "prod" }); const verified = await forge.verifyApiKey(apiKey.secret); // owner, expiry, scopes, metadata ``` * Python ```python hash = await forge.hash_password(password) # store this ok = await forge.verify_password(password, hash) token = await forge.create_session(user_id) user_id2 = await forge.validate_session(token) # str | None # one-time token: email the link, then consume it exactly once reset = await forge.create_token(user_id, "password-reset", 900, b"return=/settings") consumed = await forge.consume_token(reset, "password-reset") # TokenConsumption | None ``` * Rust ```rust use std::time::Duration; use forgelib::SessionOpts; let hash = forge.auth().hash_password(&password).await?; // store this let ok = forge.auth().verify_password(&password, &hash).await?; let token = forge.auth().create_session(&user_id, SessionOpts::default()).await?; let session = forge.auth().validate_session(token.as_str()).await?; // Option // one-time token: email the link, then consume it exactly once let reset = forge.auth().create_token_with_payload(&user_id, "password-reset", Duration::from_secs(900), b"return=/settings".as_slice().into()).await?; let consumed = forge.auth().consume_token_with_payload(reset.as_str(), "password-reset").await?; ``` Only the hash of a session token, API key, or one-time token is stored, and plaintext credentials are shown once. `validate_session` slides the idle deadline and returns nothing for an unknown, expired, or revoked token rather than erroring. After a password check passes, call `needs_rehash`; when it returns true, hash the same verified plaintext again and replace the stored PHC string in the same login flow. This is how Argon2 parameter upgrades roll out without forced password resets, and the behavior is identical in all four languages. One-time tokens carry up to 4096 opaque bytes and are consumed atomically: the first consume with the right purpose deletes the token and returns both user ID and payload; wrong-purpose, expired, unknown, and already-used tokens return nothing, and a wrong purpose leaves a live token intact. API keys may expire and carry up to 32 bounded application-owned scopes plus 4096 bytes of metadata. Verification returns owner, expiry, scopes, and metadata in one lookup. Forge stores these facts but never interprets scopes or grants permission. Delivery and identity policy stay yours. Forge does not send email or SMS and does not own user tables, organization membership, permission policy, OAuth/OIDC providers, MFA workflows, or account recovery UX. ## Rate limit [Section titled “Rate limit”](#rate-limit) An atomic check-and-consume against a limit, returning a decision carrying the IETF RateLimit header fields. Token bucket by default, or sliding window. Lineage: token bucket and GCRA, with the IETF RateLimit headers. * Node ```ts const d = await forge.rateLimitCheck("login", email, 20, 60); if (!d.allowed) { // d.retryAfterSeconds says when the client can try again } ``` * Python ```python d = await forge.rate_limit_check("login", email, 20, 60) if not d.allowed: ... # d.retry_after_seconds says when the client can try again ``` * Rust ```rust use std::time::Duration; use forgelib::Limit; let d = forge.ratelimit() .check("login", &email, Limit::per_duration(20, Duration::from_secs(60))) .await?; if !d.allowed { // d.retry_after says when the client can try again } ``` A denied request is a normal decision with `allowed: false`, not an error. By default the limiter fails open: if its backend is unavailable it allows the request and logs a warning, so an outage doesn’t lock everyone out. Turn that off (per call or in `forge.toml`) for payment- or abuse-sensitive buckets. The policy lives in the call you make, the limit and the window, not in server config. Checks may consume any positive integer cost up to the bucket capacity; a zero cost is `Invalid`, and a cost above capacity is `Limit`. Forge counts abstract units only. Common policies count requests, batch items, rounded compute seconds, model tokens, storage bytes, or a shared per-tenant budget. Currency conversion, prices, invoices, and billing records stay in the application. When the final cost is unknown, reserve a conservative upper bound for at most one hour, then commit the actual cost or release it. Commit refunds unused units. Expiry and release refund the whole reservation. Repeating the same commit or release is idempotent; changing an already committed value or crossing commit and release fails with `Precondition`, so retries cannot double-charge or drive usage negative. ## Schedule [Section titled “Schedule”](#schedule) Recurring cron jobs and one-off work at a future time. A schedule enqueues onto a normal queue when it fires, so an ordinary worker drains it. Lineage: cron, Unix `at`, and Kubernetes CronJob. Schedules do not store callbacks A schedule stores the queue name and payload to enqueue later. The code to run lives in the worker for that queue. Use one queue per task type, or put a `type` field in the payload and dispatch inside one worker. * Node ```ts // every day at midnight UTC, enqueue onto "reports" await forge.scheduleCron("nightly", "0 0 * * *", "reports", "{}"); // once, at a specific time await forge.scheduleAt(Date.now() + 3600_000, "reminders", payload); ``` * Python ```python # every day at midnight UTC, enqueue onto "reports" await forge.schedule_cron("nightly", "0 0 * * *", "reports", "{}") # once, at a specific time (epoch milliseconds) await forge.schedule_at(now_ms + 3_600_000, "reminders", payload) ``` * Rust ```rust use std::time::{Duration, SystemTime}; use forgelib::ScheduleOpts; // every day at midnight UTC, enqueue onto "reports" forge.schedule().cron("nightly", "0 0 * * *", "reports", b"{}".to_vec().into(), ScheduleOpts::new()).await?; // once, an hour from now let when = SystemTime::now() + Duration::from_secs(3600); forge.schedule().at(when, "reminders", payload.into(), ScheduleOpts::new()).await?; ``` * Go ```go // Every day at midnight UTC. Catch up at most three recent missed runs. err := forge.ScheduleCron(ctx, "nightly", "0 0 * * *", "reports", []byte("{}"), forgelib.ScheduleOptions{MisfirePolicy: forgelib.MisfireCatchUp, MaxCatchUp: 3}) // Once, at a specific instant. time.Time is normalized to UTC. jobID, err := forge.ScheduleAt(ctx, time.Now().Add(time.Hour), "reminders", payload, forgelib.ScheduleOptions{}) ``` For example, to delete an expiring link, schedule a payload onto an `expire-link` queue, run a worker for that queue, and keep a scheduler loop ticking in the background. * Node ```ts const EXPIRE_QUEUE = "expire-link"; // Producer: register future work. This does not run deleteLink yet. await forge.scheduleAt( Date.now() + 3600_000, EXPIRE_QUEUE, JSON.stringify({ slug }), // schedule payloads use the raw queue contract ); // Worker: this is the code that runs when the scheduled job is delivered. void forge.worker<{ slug: string }>(EXPIRE_QUEUE, async (job) => { await deleteLink(job.payload.slug); }); // Scheduler: turns due schedules into queue jobs. setInterval(async () => { await forge.runSchedulerOnce(); }, 30_000); ``` * Python ```python EXPIRE_QUEUE = "expire-link" # Producer: register future work. This does not run delete_link yet. await forge.schedule_at( now_ms() + 3_600_000, EXPIRE_QUEUE, json.dumps({"slug": slug}), # schedule payloads use the raw queue contract ) # Worker: this is the code that runs when the scheduled job is delivered. async def handle_expire(job): await delete_link(job.payload["slug"]) # Scheduler: turns due schedules into queue jobs. async def scheduler(): while True: await forge.run_scheduler_once() await asyncio.sleep(30) asyncio.create_task(forge.worker(EXPIRE_QUEUE, handle_expire)) asyncio.create_task(scheduler()) ``` * Rust ```rust use std::time::{Duration, SystemTime}; use forgelib::ScheduleOpts; // Producer: register future work. This does not run delete_link yet. let payload = serde_json::json!({ "slug": slug }).to_string(); let when = SystemTime::now() + Duration::from_secs(3600); forge.schedule().at(when, "expire-link", payload.into(), ScheduleOpts::new()).await?; // Worker: this is the code that runs when the scheduled job is delivered. let worker_forge = forge.clone(); tokio::spawn(async move { worker_forge.worker("expire-link").run(move |job| async move { let payload: serde_json::Value = serde_json::from_slice(&job.payload).map_err(|e| e.to_string())?; let slug = payload["slug"].as_str().ok_or("missing slug")?; delete_link(slug).await.map_err(|e| e.to_string()) }).await; }); // Scheduler: turns due schedules into queue jobs. tokio::spawn({ let forge = forge.clone(); async move { loop { if let Err(e) = forge.run_scheduler_once().await { tracing::warn!(error = %e, "scheduler tick failed"); } tokio::time::sleep(Duration::from_secs(30)).await; } } }); ``` Nothing fires until something ticks the scheduler. Run the managed loop (`run_scheduler()` in Rust) or call `run_scheduler_once()` on an interval, as shown in [Operations](/operations/#the-scheduler). Ticking is safe on every replica. Every occurrence gets a job ID derived from namespace, schedule name, and its UTC timestamp, so a retry cannot duplicate that occurrence. Cron expressions are always 5-field UTC. Absolute inputs are instants and are normalized to UTC before storage. Forge does not accept timezone names and therefore has no daylight-saving-time ambiguity; applications that need “09:00 Europe/Paris” calculate and register the desired UTC schedule themselves. Misfires are explicit per schedule. `run_once` is the default and dispatches only the most recent missed occurrence. `skip` drops missed occurrences. `catch_up` dispatches the most recent requested count in chronological order, with a hard maximum of 100, then advances to the next future occurrence. This cap applies after any length of downtime and prevents restart storms. One-shots use the same policy but can dispatch at most once. Named schedules can be inspected, paused, and resumed without deleting their configuration. Paused entries are excluded from due count and lag. Scheduler diagnostics report the oldest due lag, current due count, latest successful tick in UTC epoch milliseconds, and cumulative enqueue failures. Failed enqueue leaves the occurrence due for the next pass; successful or skipped occurrences advance atomically. These APIs schedule queue deliveries only. Dependencies, DAGs, compensation, human approval, and multi-step workflow state remain application concerns. ## Config and flags [Section titled “Config and flags”](#config-and-flags) Runtime settings and typed feature flags. Flags can be always on, always off, a percentage rollout, an allow-list, or a static JSON value with an application-defined stable variant. Lineage: 12-factor config and OpenFeature. * Node ```ts await forge.setFlagPercent("new-ui", 25); const on = await forge.flag("new-ui", false, userId); await forge.configSet("max_upload_mb", "25"); const raw = await forge.configGet("max_upload_mb"); // string | null ``` * Python ```python await forge.set_flag_percent("new-ui", 25) on = await forge.flag("new-ui", False, user_id) await forge.config_set("max_upload_mb", "25") raw = await forge.config_get("max_upload_mb") # str | None ``` * Rust ```rust use forgelib::{FlagRule, EvalCtx}; forge.config().set_flag("new-ui", FlagRule::Percent(25)).await?; let on = forge.config().flag("new-ui", false, &EvalCtx::user(&user_id)).await; forge.config().set_raw("max_upload_mb", "25").await?; let raw = forge.config().get_raw("max_upload_mb").await?; // Option ``` * Go ```go err := forgeClient.SetFlag(ctx, "new-ui", forge.FlagRule{Kind: forge.FlagPercent, Percent: 25}) if err != nil { return err } on := forgeClient.Flag(ctx, "new-ui", false, userID) err = forgeClient.ConfigSet(ctx, "max_upload_mb", []byte("25")) raw, err := forgeClient.ConfigGet(ctx, "max_upload_mb") ``` `flag` never throws. If the flag is missing or the backend is down it returns the default you passed and records why, so a flag check cannot take down a request. `flag_details` adds canonical typed JSON, a stable variant, a stable reason, and an error category. Forge does not collect exposure or experiment analytics; the application decides consent, identity, sampling, and destination. The official provider adapters use the official OpenFeature SDK in every supported language: Rust enables the crate’s `openfeature` feature and uses `forgelib::openfeature::ForgeProvider`; Node imports `ForgeProvider` from `forgelib/openfeature`; Python installs `forgelib[openfeature]` and imports `forgelib.openfeature.ForgeProvider`; Go imports `github.com/isala404/forge/bindings/go/openfeatureprovider`. Providers register no global hooks or evaluation context. OpenFeature owns hook ordering and passes merged invocation-local context to hooks. Each adapter exposes an application-scoped telemetry hook; the Node, Python, and Go adapters wrap the official OpenFeature OpenTelemetry hooks, while Rust’s hook emits `feature_flag.evaluation` tracing events with `feature_flag.key`, provider, result value, stable variant/reason, context ID, and `error.type` where applicable. ```rust use forgelib::openfeature::{ForgeProvider, OpenTelemetryHook}; use open_feature::OpenFeature; let mut api = OpenFeature::default(); api.set_provider(ForgeProvider::new(forge.clone())).await; api.add_hook(OpenTelemetryHook).await; // application chooses this global scope let client = api.create_client(); ``` Use `get_many_raw` / `configGetMany` / `config_get_many` / `ConfigGetMany` and `flag_details_many` / `flagDetailsMany` / `flag_details_many` / `FlagDetailsMany` for startup reads. Each accepts at most 256 exact keys, preserves request order and duplicates, and the PostgreSQL implementation uses one query per bulk operation instead of one query per key. Snapshots capture only explicitly requested config values and already-evaluated flag requests. They are read-only, expire after 1–86,400 seconds, reject out-of-scope reads, cap their portable JSON form at 1 MiB, and never re-evaluate flags offline. `no_secrets` is an application assertion that captured config contains no secrets; `application_protected` means the application must encrypt and authenticate the encoded snapshot before it leaves the trusted server boundary. Forge never infers which config values are secret and never extends a stale snapshot. Reads are cached in-process for about 30 seconds, so a change is visible everywhere within that window rather than instantly. An environment variable `FORGE_CFG_` overrides a stored config value, which is useful for an ops override. # Quickstart > Add Forge to a backend, configure PostgreSQL, and make the first calls in Rust, JavaScript, Python, or Go. Add Forge to the backend process, point it at a PostgreSQL database it can own, and call `init()` to get one client for all eight primitives. Forge does not create an HTTP server, install process signal handlers, or add frontend dependencies. ## Install [Section titled “Install”](#install) * Node ```sh npm install forgelib ``` * Python ```sh pip install forgelib ``` * Rust ```sh cargo add forgelib ``` * Go ```sh go get github.com/isala404/forge/bindings/go ``` ## Give Forge a database [Section titled “Give Forge a database”](#give-forge-a-database) Forge keeps its own state in Postgres. It owns its `forge_*` tables, so give it a database (or its own schema) and keep your product tables separate. Development startup migrates automatically; production deployments call the migration API before starting the application. ```sh # any Postgres works; create one for Forge to own createdb myapp ``` Tip No Postgres handy? Set the backend to `memory` and Forge runs entirely in-process, which is how the test suite runs. See [Configuration](/configuration/#backends). ## Write forge.toml [Section titled “Write forge.toml”](#write-forgetoml) Configuration lives in a `forge.toml` at your project root. It’s the single source of config: `init()` reads it and builds the runtime. String values can read the environment with `${VAR}` or `${VAR:-default}`, so the same file works in tests and in production. forge.toml ```toml [postgres] url = "${DATABASE_URL:-postgres://localhost/myapp}" ``` Everything else has a convenient development default, so this is enough to start locally. Set `forge.environment = "production"` in production; automatic migration then defaults off. The full list of settings is in [Configuration](/configuration/). ## Make your first calls [Section titled “Make your first calls”](#make-your-first-calls) `init()` reads `forge.toml`, migrates the system database, and returns a client. The same calls look like this in each language. (Pick a language once and the tabs on this site follow your choice.) * Node ```ts import { ForgeClient } from "forgelib"; const forge = await ForgeClient.init(); // reads ./forge.toml, runs migrations // auth: hash a password, open a session const hash = await forge.hashPassword(password); const token = await forge.createSession(userId); // key/value: an atomic counter const views = await forge.kvIncr(`views:${userId}`, 1); // queue: hand work to the background await forge.queue<{ to: string }>("emails").enqueue({ to: email }); // rate limit: 20 attempts per minute, keyed by email const limit = await forge.rateLimitCheck("login", email, 20, 60); if (!limit.allowed) throw new Error("slow down"); ``` * Python ```python import forgelib forge = await forgelib.ForgeClient.init() # reads ./forge.toml, runs migrations # auth: hash a password, open a session hash = await forge.hash_password(password) token = await forge.create_session(user_id) # key/value: an atomic counter views = await forge.kv_incr(f"views:{user_id}", 1) # queue: hand work to the background await forge.queue("emails").enqueue({"to": email}) # rate limit: 20 attempts per minute, keyed by email limit = await forge.rate_limit_check("login", email, 20, 60) if not limit.allowed: raise RuntimeError("slow down") ``` * Rust ```rust use std::time::Duration; use forgelib::{Forge, SessionOpts, EnqueueOpts, Limit}; let forge = Forge::init().await?; // reads ./forge.toml, runs migrations // auth: hash a password, open a session let hash = forge.auth().hash_password(&password).await?; let token = forge.auth().create_session(&user_id, SessionOpts::default()).await?; // key/value: an atomic counter let views = forge.kv().incr(&format!("views:{user_id}"), 1).await?; // queue: hand work to the background forge.queue().enqueue("emails", payload.into(), EnqueueOpts::new()).await?; // rate limit: 20 attempts per minute, keyed by email let limit = forge.ratelimit() .check("login", &email, Limit::per_duration(20, Duration::from_secs(60))) .await?; if !limit.allowed { /* return 429 */ } ``` * Go ```go import ( "context" "time" forge "github.com/isala404/forge/bindings/go" ) client, err := forge.InitFrom(context.Background(), "forge.toml") if err != nil { return err } defer client.Close(context.Background()) hash, err := client.HashPassword(context.Background(), password) if err != nil { return err } token, err := client.CreateSession(context.Background(), userID, forge.SessionOptions{}) if err != nil { return err } views, err := client.KVIncr(context.Background(), "views:"+userID, 1) if err != nil { return err } _, err = client.Enqueue(context.Background(), "emails", payload, forge.EnqueueOptions{}) if err != nil { return err } limit, err := client.RateLimitCheck(context.Background(), "login", email, forge.RateLimitOptions{Max: 20, Per: time.Minute}) if err != nil { return err } if !limit.Allowed { /* return 429 */ } ``` One shape difference is worth noting. Rust reaches each primitive through an accessor and uses option structs, like `forge.kv().set(key, value, SetOpts::new())`. JavaScript and Python keep the exact raw methods on the client (`forge.kvSet(...)`, `forge.kv_set(...)`) for contract parity and add native JSON handles for application values. Go uses one concrete client with explicit option structs and `context.Context` on every blocking operation. ## Run the background work [Section titled “Run the background work”](#run-the-background-work) Enqueued jobs sit in the queue until an application-owned worker process starts a Forge worker. Schedules fire only when the application runs a scheduler loop. A normal deployment runs workers for the queues it owns, a scheduler loop, and periodic maintenance. [Operations](/operations/) covers those roles, and [Integrations](/integrations/) shows HTTP, deployment, MCP, and observability boundaries. ## Next steps [Section titled “Next steps”](#next-steps) [Primitives](/primitives/)What each of the eight does, and the methods you'll actually call. [Integrations](/integrations/)HTTP lifecycle, workers, deployment, MCP, CloudEvents, and observability. [Configuration](/configuration/)Every forge.toml setting and its default. [Operations](/operations/)Workers, the scheduler, health, and going to production. # Recipes > End-to-end patterns built from the example apps: auth flows, background jobs, live updates, file handling, and scheduled work. These are the patterns that show up in the [canonical examples](https://github.com/isala404/forge/tree/main/examples). Each language has one application example instead of parallel copies of every app. The code here shows all languages where comparison helps, trimmed to the Forge calls so the shape is easy to lift into your own handlers. ## Production recipe index [Section titled “Production recipe index”](#production-recipe-index) The deployment recipes live beside their operational guarantees: [migration jobs](/operations/#deploy-migrations-before-the-application), [managed workers and graceful shutdown](/operations/#workers), [transactional outbox delivery](/operations/#transactional-outbox), [liveness and readiness endpoints](/operations/#health-checks), and [Prometheus export](/operations/#metrics-and-tracing). Queue producers should supply a deterministic job ID for a business operation and make handlers safe to repeat; use the outbox when enqueue must agree with an application-owned SQL transaction. For blob uploads, use the Forge proxy signature when the server must enforce a hard body limit or application-specific authorization, and native S3 presigning when direct provider transfer is acceptable. ## Password auth [Section titled “Password auth”](#password-auth) Email and password auth is about thirty lines of Forge calls per handler: hash on the way in, verify on the way out, and issue sessions you can revoke. There’s no separate users table to wire up first: `kv` stores the user record and `auth` owns the hashing and the sessions. Know the boundary This is the “email and password, done right” baseline. Password reset, email verification, and magic links are covered by one-time tokens (below); Forge mints and checks the token, your app sends the email. OAuth and 2FA are their own product surface — reach for a dedicated auth provider once you need those rather than growing this into one. Sign-up does four things: throttle by email, hash the password, claim the email atomically, and open a session. Writing the email key with “set if not present” makes it its own uniqueness check, so there’s no read-then-write race. * Node ```ts // throttle by email, fail open so a limiter outage doesn't lock people out const limit = await forge.rateLimitCheck("auth", email, 20, 60, true); if (!limit.allowed) return c.json({ error: "slow down" }, 429); const user = { id: randomUUID(), email, passwordHash: await forge.hashPassword(password) }; const body = JSON.stringify(user); // the email key is its own uniqueness guard: false means already taken const created = await forge.kvSet(`user:email:${email}`, body, null, true); if (!created) return c.json({ error: "email already registered" }, 409); await forge.kvSet(`user:id:${user.id}`, body); const token = await forge.createSession(user.id, 86_400, 30 * 86_400); ``` * Python ```python # throttle by email, fail open so a limiter outage doesn't lock people out limit = await forge.rate_limit_check("auth", email, 20, 60.0, True) if not limit.allowed: raise HTTPException(429, "slow down") user = {"id": uuid4().hex, "email": email, "password_hash": await forge.hash_password(password)} body = json.dumps(user) # the email key is its own uniqueness guard: False means already taken created = await forge.kv_set(f"user:email:{email}", body, None, True) if not created: raise HTTPException(409, "email already registered") await forge.kv_set(f"user:id:{user['id']}", body) token = await forge.create_session(user["id"], 86_400, 30 * 86_400) ``` * Rust ```rust use std::time::Duration; use forgelib::{Bytes, Limit, FailMode, SetOpts, SetMode, SessionOpts}; // throttle by email, fail open so a limiter outage doesn't lock people out let limit = forge.ratelimit() .check_with("auth", &email, Limit::per_duration(20, Duration::from_secs(60)), FailMode::Open) .await?; if !limit.allowed { return too_many_requests(); } let user = User { id: new_id(), email: email.clone(), password_hash: forge.auth().hash_password(&password).await?.as_str().to_string() }; let body = Bytes::from(serde_json::to_vec(&user)?); // the email key is its own uniqueness guard: false means already taken let created = forge.kv() .set(&format!("user:email:{email}"), body.clone(), SetOpts::new().with_mode(SetMode::IfNotExists)).await?; if !created { return conflict("email already registered"); } forge.kv().set(&format!("user:id:{}", user.id), body, SetOpts::new()).await?; let token = forge.auth().create_session(&user.id, SessionOpts::new() .with_idle_timeout(Duration::from_secs(86_400)) .with_absolute_timeout(Duration::from_secs(30 * 86_400))).await?; ``` Logging in runs the same throttle, verifies the password against the stored hash, then opens a fresh session. Keep the rate-limit check ahead of `verifyPassword` so a stuffing run can’t burn CPU on hashing. * Node ```ts const limit = await forge.rateLimitCheck("auth", email, 20, 60, true); if (!limit.allowed) return c.json({ error: "slow down" }, 429); const stored = await forge.kvGet(`user:email:${email}`); if (!stored) return c.json({ error: "invalid email or password" }, 401); const user = JSON.parse(stored); // the throttle runs before verifyPassword, so credential stuffing can't burn CPU on hashing if (!(await forge.verifyPassword(password, user.passwordHash))) return c.json({ error: "invalid email or password" }, 401); const token = await forge.createSession(user.id, 86_400, 30 * 86_400); ``` * Python ```python limit = await forge.rate_limit_check("auth", email, 20, 60.0, True) if not limit.allowed: raise HTTPException(429, "slow down") stored = await forge.kv_get(f"user:email:{email}") if not stored: raise HTTPException(401, "invalid email or password") user = json.loads(stored) # the throttle runs before verify_password, so credential stuffing can't burn CPU on hashing if not await forge.verify_password(password, user["password_hash"]): raise HTTPException(401, "invalid email or password") token = await forge.create_session(user["id"], 86_400, 30 * 86_400) ``` * Rust ```rust use forgelib::PhcString; let limit = forge.ratelimit() .check_with("auth", &email, Limit::per_duration(20, Duration::from_secs(60)), FailMode::Open) .await?; if !limit.allowed { return too_many_requests(); } let Some(bytes) = forge.kv().get(&format!("user:email:{email}")).await? else { return invalid_login(); }; let user: User = serde_json::from_slice(&bytes)?; // the throttle runs before verify_password, so credential stuffing can't burn CPU on hashing if !forge.auth().verify_password(&password, &PhcString::new(user.password_hash.clone())).await? { return invalid_login(); } let token = forge.auth().create_session(&user.id, SessionOpts::new() .with_idle_timeout(Duration::from_secs(86_400)) .with_absolute_timeout(Duration::from_secs(30 * 86_400))).await?; ``` Every protected route resolves the caller’s token to a user (or 401s), and logout revokes the session so the token stops working immediately. * Node ```ts // middleware: resolve the bearer token (or session cookie) to a user id, else 401 const userId = await forge.validateSession(bearerToken(c.req.header("authorization"))); if (!userId) return c.json({ error: "authentication required" }, 401); // logout: revoke the session so the token stops resolving await forge.revokeSession(bearerToken(c.req.header("authorization"))); ``` * Python ```python # middleware: resolve the bearer token (or session cookie) to a user id, else 401 user_id = await forge.validate_session(bearer) if not user_id: raise HTTPException(401, "authentication required") # logout: revoke the session so the token stops resolving await forge.revoke_session(bearer) ``` * Rust ```rust // middleware: resolve the bearer token (or session cookie) to a session, else 401 let Some(session) = forge.auth().validate_session(bearer).await? else { return unauthorized(); }; let user_id = session.user_id; // logout: revoke the session so the token stops resolving forge.auth().revoke_session(bearer).await?; ``` Password reset is a one-time token plus an email your app sends. Mint a short-lived token scoped to `"password-reset"`, put it in the link, and consume it when the user comes back; the first consume wins and every retry after that returns nothing. Respond identically whether or not the email exists so the endpoint doesn’t confirm accounts. * Node ```ts // request handler: always answer 200; only mint + email when the account exists const found = await forge.kvGet(`user:email:${email}`); if (found) { const user = JSON.parse(found); const token = await forge.createToken(user.id, "password-reset", 900); // 15 min await forge.queue("emails").enqueue({ to: email, resetLink: linkFor(token) }); } // reset handler: consume exactly once, then re-hash and revoke every session const consumed = await forge.consumeToken(token, "password-reset"); if (!consumed) return c.json({ error: "link expired" }, 400); await updatePasswordHash(consumed.userId, await forge.hashPassword(newPassword)); await forge.revokeAllSessions(consumed.userId); ``` * Python ```python # request handler: always answer 200; only mint + email when the account exists found = await forge.kv_get(f"user:email:{email}") if found: user = json.loads(found) token = await forge.create_token(user["id"], "password-reset", 900) # 15 min await forge.queue("emails").enqueue({"to": email, "reset_link": link_for(token)}) # reset handler: consume exactly once, then re-hash and revoke every session consumed = await forge.consume_token(token, "password-reset") if not consumed: raise HTTPException(400, "link expired") await update_password_hash(consumed.user_id, await forge.hash_password(new_password)) await forge.revoke_all_sessions(consumed.user_id) ``` * Rust ```rust // request handler: always answer 200; only mint + email when the account exists if let Some(bytes) = forge.kv().get(&format!("user:email:{email}")).await? { let user: User = serde_json::from_slice(&bytes)?; let token = forge.auth() .create_token(&user.id, "password-reset", Duration::from_secs(900)) // 15 min .await?; forge.queue().enqueue("emails", email_payload(&email, token.as_str()), EnqueueOpts::default()).await?; } // reset handler: consume exactly once, then re-hash and revoke every session let Some(consumed) = forge.auth().consume_token_with_payload(token, "password-reset").await? else { return link_expired(); }; update_password_hash(&consumed.user_id, forge.auth().hash_password(&new_password).await?).await?; forge.auth().revoke_all_sessions(&consumed.user_id).await?; ``` The same shape does email verification (`"email-verify"` purpose, consume on the confirmation link) and magic-link login (consume, then `create_session` for the returned user id). The purpose string is what stops a reset token from being replayed against the verify endpoint. ## Offload slow work to a queue [Section titled “Offload slow work to a queue”](#offload-slow-work-to-a-queue) Keep the request fast by doing only the cheap part inline and handing the rest to a queue. The URL shortener bumps a click counter on the redirect and queues everything else, so the user gets their redirect immediately. * Node ```ts // in the redirect handler: cheap now, the rest later await forge.kvIncr(`clicks:${slug}`, 1); await forge.queue<{ slug: string }>("clicks").enqueue({ slug }, { maxAttempts: 3 }); ``` * Python ```python # in the redirect handler: cheap now, the rest later await forge.kv_incr(f"clicks:{slug}", 1) await forge.queue("clicks").enqueue({"slug": slug}, max_attempts=3) ``` * Rust ```rust use forgelib::{Bytes, EnqueueOpts}; use serde_json::json; // in the redirect handler: cheap now, the rest later forge.kv().incr(&format!("clicks:{slug}"), 1).await?; forge.queue().enqueue("clicks", Bytes::from(json!({ "slug": slug }).to_string()), EnqueueOpts::new().with_max_attempts(3)).await?; ``` A worker leases jobs, does the work, and acks on success or nacks on failure so the job is retried. Run one of these per queue (see [Operations](/operations/#workers) for a managed version). * Node ```ts await forge.worker<{ slug: string }>("clicks", async (job) => { await handle(forge, job.payload); }); ``` * Python ```python await forge.worker("clicks", lambda job: handle(forge, job.payload)) ``` * Rust ```rust use forgelib::{DequeueOpts, NackOpts}; loop { let Some(job) = forge.queue().dequeue("clicks", DequeueOpts::new()).await? else { continue }; match handle(&forge, &job).await { Ok(()) => { forge.queue().ack(&job).await?; } Err(_) => { forge.queue().nack(&job, NackOpts::default()).await?; } } } ``` ## Publish backend events through an application transport [Section titled “Publish backend events through an application transport”](#publish-backend-events-through-an-application-transport) The worker that drains the clicks queue publishes a bounded invalidation hint after the durable count changes. An application-owned SSE route subscribes to that topic and forwards each hint to a connected client, then drops the subscription when the client disconnects. The client refetches the count; it never treats the hint as authoritative state. WebSockets or another transport can use the same backend pattern. * Node ```ts // in the worker handler, after committing the new count await forge.pubsubPublish(`clicks:${slug}`, encodeInvalidationEvent({ schema_version: 1, tags: [`links/${slug}`], query_keys: [["link", slug]], revision: String(total), }).toString("utf8")); // in the SSE route (Hono) return streamSSE(c, async (stream) => { const events = await forge.topic(`clicks:${slug}`) .subscribe(); stream.onAbort(() => { void events.return?.(); }); for await (const event of events) { await stream.writeSSE({ data: JSON.stringify(event) }); } }); ``` * Python ```python # in the worker handler, after committing the new count hint = encode_invalidation_event({ "schema_version": 1, "tags": [f"links/{slug}"], "query_keys": [["link", slug]], "revision": str(total), }) await forge.pubsub_publish(f"clicks:{slug}", hint.decode()) # in the SSE route (FastAPI) async def live(slug: str): topic = forge.topic(f"clicks:{slug}") async def events(): async for event in topic.subscribe(): yield f"data: {json.dumps(event)}\n\n" return StreamingResponse(events(), media_type="text/event-stream") ``` * Rust ```rust use futures_util::StreamExt; // in the worker handler, after committing the new count let hint = InvalidationEvent::new( vec![format!("links/{slug}")], vec![vec![json!("link"), json!(slug)]], Some(total.to_string()), )?; forge.pubsub().publish(&format!("clicks:{slug}"), Bytes::from(hint.encode()?)).await?; // in the SSE route let mut sub = forge.pubsub().subscribe(&format!("clicks:{slug}")).await?; while let Some(Ok(payload)) = sub.next().await { // write `payload` to the response as an SSE `data:` line } ``` Note Pub/sub only reaches connected clients. The durable number lives in the key/value counter; the topic only tells open consumers to refresh. See [Event delivery](/event-delivery/) for transport limits and client ownership. ## Store and serve a file [Section titled “Store and serve a file”](#store-and-serve-a-file) Put generated or uploaded bytes under a key with a content type, read them back to serve them, or hand the client a presigned link so it can fetch the file directly. * Node ```ts await forge.blobPut(`qr/${slug}.svg`, svg, "image/svg+xml"); const stored = await forge.blobGet(`qr/${slug}.svg`); // string | null const url = (await forge.blobPresignDownload(`qr/${slug}.svg`, 3600)).url; ``` * Python ```python await forge.blob_put(f"qr/{slug}.svg", svg, "image/svg+xml") # svg is bytes stored = await forge.blob_get(f"qr/{slug}.svg") # bytes | None url = (await forge.blob_presign_download(f"qr/{slug}.svg", 3600)).url ``` * Rust ```rust use std::time::Duration; use forgelib::{Bytes, PutOpts}; forge.blob().put(&format!("qr/{slug}.svg"), Bytes::from(svg), PutOpts::new().with_content_type("image/svg+xml")).await?; let svg = forge.blob().get(&format!("qr/{slug}.svg")).await?; // Option let url = forge.blob() .presign_download(&format!("qr/{slug}.svg"), Duration::from_secs(3600)).await?.url; ``` Presigned links need a `signing_secret` set in `forge.toml`. For uploads, hand back the `.url` from the structured proxy presign result; the application proxy verifies its method, key, expiry, namespace, and `max_bytes` before accepting the body. ## Run work later [Section titled “Run work later”](#run-work-later) Schedule a one-shot for delayed work, like expiring a link at its TTL, or a cron for recurring jobs. Both land on a normal queue, so the same workers drain them. * Node ```ts // expire this link in an hour await forge.scheduleAt(Date.now() + 3600_000, "expire", JSON.stringify({ slug })); // a daily report await forge.scheduleCron("nightly-report", "0 0 * * *", "reports", "{}"); ``` * Python ```python # expire this link in an hour await forge.schedule_at(now_ms + 3_600_000, "expire", json.dumps({"slug": slug})) # a daily report await forge.schedule_cron("nightly-report", "0 0 * * *", "reports", "{}") ``` * Rust ```rust use std::time::{Duration, SystemTime}; use forgelib::ScheduleOpts; // expire this link in an hour let when = SystemTime::now() + Duration::from_secs(3600); forge.schedule().at(when, "expire", payload.into(), ScheduleOpts::new()).await?; // a daily report forge.schedule().cron("nightly-report", "0 0 * * *", "reports", b"{}".to_vec().into(), ScheduleOpts::new()).await?; ``` Schedules only fire when the scheduler is ticked. The same loop that ticks it is a good place to run Forge’s maintenance sweep. See [Operations](/operations/#the-scheduler) for the loop these apps run on every replica. # Reference > Cross-language method index, return types, the error taxonomy, and limits. Here is the surface in one place. Rust groups methods under accessors and takes option structs. Node and Python put methods directly on the client with plain arguments. Go uses methods on its `Forge` handle. Node and Python also expose native JSON handles from the main import (`forge.queue("emails")`, `forge.queue("emails")`) so application payloads stay typed without a separate overlay. ## Creating a client [Section titled “Creating a client”](#creating-a-client) * Node ```ts import { ForgeClient } from "forgelib"; const forge = await ForgeClient.init(); // reads ./forge.toml const forge2 = await ForgeClient.initFrom("path/forge.toml"); ``` * Python ```python import forgelib forge = await forgelib.ForgeClient.init() # reads ./forge.toml forge = await forgelib.ForgeClient.init_from("path/forge.toml") ``` * Rust ```rust let forge = forgelib::Forge::init().await?; // reads ./forge.toml let forge = forgelib::Forge::init_from("path/forge.toml").await?; let forge = forgelib::Forge::init_from_str(toml).await?; // inject your own backend for one primitive, leave the rest on Postgres let forge = forgelib::Forge::builder().kv(custom_kv).build().await?; ``` * Go ```go forgeClient, err := forge.InitDefault(ctx) // reads ./forge.toml forgeClient2, err := forge.InitFrom(ctx, "path/forge.toml") ``` ## Deterministic test clients [Section titled “Deterministic test clients”](#deterministic-test-clients) The memory test factories inject a manual clock and seeded, non-cryptographic entropy. Advancing time drives KV TTLs, sessions and tokens, delayed jobs, retry availability, rate-limit refill, and schedules without sleeping. Never use seeded entropy for production credentials. * Node ```ts const config = '[forge]\nmode = "memory"\nenvironment = "test"\n'; const forge = await ForgeClient.initMemoryForTesting(config, Date.UTC(2026, 0, 1), 42); forge.advanceTestClock(60); // seconds ``` * Python ```python config = '[forge]\nmode = "memory"\nenvironment = "test"\n' forge = await forgelib.ForgeClient.init_memory_for_testing(config, 1_767_225_600_000, 42) forge.advance_test_clock(60.0) # seconds ``` * Rust ```rust let forge = forgelib::Forge::init_memory_for_testing(config, SystemTime::UNIX_EPOCH, 42).await?; forge.advance_test_clock(Duration::from_secs(60))?; ``` * Go ```go clock := forge.NewManualClock(time.Unix(0, 0)) client, err := forge.NewMemoryForTesting(config, forge.TestOptions{ ManualClock: clock, Random: forge.NewSeededReader(42), }) if err != nil { return err } if err := client.AdvanceTestClock(time.Minute); err != nil { return err } ``` Rust’s `testing::TestDatabase::new()` creates and later drops one isolated PostgreSQL database using the maintenance URL in `TEST_DATABASE_URL`; integration tests never reuse an application database. The repository’s Node, Python, and Go conformance runners follow the same per-run isolation model. The CI integration job sets a sentinel that turns a missing database URL into failure rather than a skipped pass. ## Methods by primitive [Section titled “Methods by primitive”](#methods-by-primitive) ### Key/value [Section titled “Key/value”](#keyvalue) | Operation | Rust | Node | Python | Go | | ----------------- | ----------------------- | ---------------------- | ------------------------- | ------------------ | | Get | `kv().get` | `kvGet` | `kv_get` | `KVGet` | | Set (TTL, NX, XX) | `kv().set` | `kvSet` | `kv_set` | `KVSet` | | Get many | `kv().mget` | `kvMget` | `kv_mget` | `KVMGet` | | Increment | `kv().incr` | `kvIncr` | `kv_incr` | `KVIncr` | | Delete | `kv().delete` | `kvDelete` | `kv_delete` | `KVDelete` | | Exists | `kv().exists` | `kvExists` | `kv_exists` | `KVExists` | | Set TTL | `kv().expire` | `kvExpire` | `kv_expire` | `KVExpire` | | Compare and swap | `kv().compare_and_swap` | `kvCompareAndSwap` | `kv_compare_and_swap` | `KVCompareAndSwap` | | Prefix scan | `kv().scan` | `kvScan`, `kvScanPage` | `kv_scan`, `kv_scan_page` | `KVScan` | ### Queue [Section titled “Queue”](#queue) | Operation | Rust | Node | Python | Go | | ------------------------- | ------------------------------ | --------------------------- | ----------------------------- | --------------------------- | | Enqueue | `queue().enqueue` | `queueEnqueue` | `queue_enqueue` | `Enqueue` | | Batch enqueue | `queue().enqueue_batch` | `queueEnqueueBatch` | `queue_enqueue_batch` | `EnqueueBatch` | | Lease a job | `queue().dequeue` | `queueDequeue` | `queue_dequeue` | `Dequeue` | | Batch lease | `queue().dequeue_batch` | `queueDequeueBatch` | `queue_dequeue_batch` | `DequeueBatch` | | Ack | `queue().ack` | `queueAck` | `queue_ack` | `Ack` | | Nack | `queue().nack` | `queueNack` | `queue_nack` | `Nack` | | Extend the lease | `queue().heartbeat` | `queueHeartbeat` | `queue_heartbeat` | `Heartbeat` | | Depth and oldest age | `queue().depth` | `queueDepth` | `queue_depth` | `Depth` | | Pause / resume | `queue().pause`, `resume` | `queuePause`, `queueResume` | `queue_pause`, `queue_resume` | `PauseQueue`, `ResumeQueue` | | Throughput stats | `queue().stats` | `queueStats` | `queue_stats` | `QueueStats` | | Inspect dead letters | `queue().dead_letters` | `queueDeadLetters` | `queue_dead_letters` | `DeadLetters` | | Redrive | `queue().redrive` | `queueRedrive` | `queue_redrive` | `Redrive` | | Purge dry-run / confirmed | `queue().purge_dead_letters_*` | `queuePurgeDeadLetters*` | `queue_purge_dead_letters*` | `PurgeDeadLetters*` | | Managed worker | `worker(name)` | `worker(name, handler)` | `worker(name, handler)` | `RunWorker` | | Outbox relay pass | `run_outbox_once` | `runOutboxOnce` | `run_outbox_once` | `RunOutboxOnce` | | Continuous outbox relay | `run_outbox_relay_until` | `runOutboxRelay` | `run_outbox_relay` | `RunOutboxRelay` | ### Pub/sub [Section titled “Pub/sub”](#pubsub) | Operation | Rust | Node | Python | Go | | ------------ | ---------------------- | ----------------- | ------------------ | --------------- | | Publish | `pubsub().publish` | `pubsubPublish` | `pubsub_publish` | `Publish` | | Subscribe | `pubsub().subscribe` | `pubsubSubscribe` | `pubsub_subscribe` | `Subscribe` | | Channel name | `pubsub().channel_for` | `pubsubChannel` | `pubsub_channel` | `PubsubChannel` | ### Blob [Section titled “Blob”](#blob) | Operation | Rust | Node | Python | Go | | ---------------- | ------------------------- | -------------------------- | ----------------------------- | --------------------- | | Put | `blob().put` | `blobPut`, `blobPutObject` | `blob_put`, `blob_put_object` | `BlobPut` | | Get | `blob().get` | `blobGet` | `blob_get` | `BlobGet`, `BlobOpen` | | Head (metadata) | `blob().head` | `blobHead` | `blob_head` | `BlobHead` | | Delete | `blob().delete` | `blobDelete` | `blob_delete` | `BlobDelete` | | List | `blob().list` | `blobList` | `blob_list` | `BlobList` | | Presign download | `blob().presign_download` | `blobPresignDownload` | `blob_presign_download` | `BlobPresignDownload` | | Presign upload | `blob().presign_upload` | `blobPresignUpload` | `blob_presign_upload` | `BlobPresignUpload` | ### Auth [Section titled “Auth”](#auth) | Operation | Rust | Node | Python | Go | | ---------------------- | ------------------------- | ----------------- | ------------------ | ----------------- | | Hash password | `auth().hash_password` | `hashPassword` | `hash_password` | `HashPassword` | | Verify password | `auth().verify_password` | `verifyPassword` | `verify_password` | `VerifyPassword` | | Needs rehash | `auth().needs_rehash` | `needsRehash` | `needs_rehash` | `NeedsRehash` | | Create session | `auth().create_session` | `createSession` | `create_session` | `CreateSession` | | Validate session | `auth().validate_session` | `validateSession` | `validate_session` | `ValidateSession` | | Revoke session | `auth().revoke_session` | `revokeSession` | `revoke_session` | `RevokeSession` | | Create API key | `auth().create_api_key` | `createApiKey` | `create_api_key` | `CreateAPIKey` | | Verify API key | `auth().verify_api_key` | `verifyApiKey` | `verify_api_key` | `VerifyAPIKey` | | Create one-time token | `auth().create_token` | `createToken` | `create_token` | `CreateToken` | | Consume one-time token | `auth().consume_token` | `consumeToken` | `consume_token` | `ConsumeToken` | ### Rate limit [Section titled “Rate limit”](#rate-limit) | Operation | Rust | Node | Python | Go | | ----------------- | --------------------------------- | ---------------- | ------------------ | ---------------- | | Check and consume | `ratelimit().check`, `check_with` | `rateLimitCheck` | `rate_limit_check` | `RateLimitCheck` | ### Schedule [Section titled “Schedule”](#schedule) | Operation | Rust | Node | Python | Go | | --------- | -------------------- | ------------------ | -------------------- | ------------------ | | Cron | `schedule().cron` | `scheduleCron` | `schedule_cron` | `ScheduleCron` | | One-off | `schedule().at` | `scheduleAt` | `schedule_at` | `ScheduleAt` | | Cancel | `schedule().cancel` | `scheduleCancel` | `schedule_cancel` | `ScheduleCancel` | | List | `schedule().list` | `scheduleList` | `schedule_list` | `ScheduleList` | | Tick once | `run_scheduler_once` | `runSchedulerOnce` | `run_scheduler_once` | `RunSchedulerOnce` | ### Config and flags [Section titled “Config and flags”](#config-and-flags) | Operation | Rust | Node | Python | Go | | --------------- | ---------------------- | --------------------------------------------------------------- | ------------------------------------------------------------------------ | -------------- | | Get value | `config().get_raw` | `configGet` | `config_get` | `ConfigGet` | | Set value | `config().set_raw` | `configSet` | `config_set` | `ConfigSet` | | Delete value | `config().delete_raw` | `configDelete` | `config_delete` | `ConfigDelete` | | Evaluate a flag | `config().flag` | `flag` | `flag` | `Flag` | | Set a flag | `config().set_flag` | `setFlagPercent`, `setFlagOn`, `setFlagOff`, `setFlagAllowList` | `set_flag_percent`, `set_flag_on`, `set_flag_off`, `set_flag_allow_list` | `SetFlag` | | Delete a flag | `config().delete_flag` | `deleteFlag` | `delete_flag` | `DeleteFlag` | ### Client [Section titled “Client”](#client) | Operation | Rust | Node | Python | Go | | -------------------- | ---------------------- | --------------------- | ---------------------- | --------------------- | | Backend capabilities | `backend_capabilities` | `backendCapabilities` | `backend_capabilities` | `BackendCapabilities` | | Liveness | `is_live` | `isLive` | `is_live` | `IsLive` | | Readiness probe | `probe` | `probe` | `probe` | `Probe` | | Metrics snapshot | `metrics_snapshot` | `metricsSnapshot` | `metrics_snapshot` | `MetricsSnapshot` | | Prometheus text | `render_prometheus` | `renderPrometheus` | `render_prometheus` | `RenderPrometheus` | | Scheduler loop | `run_scheduler` | (interval) | (interval) | (interval) | | Maintenance sweep | `maintain` | `maintain` | `maintain` | `Maintain` | ### Stateless interoperability helpers [Section titled “Stateless interoperability helpers”](#stateless-interoperability-helpers) These functions need no initialized client and add no transport or deployment behavior. See [Integrations](/integrations/) for their bounded profile and recipes. | Operation | Rust | Node | Python | Go | | -------------------------------------- | ------------------------------------------ | ------------------ | -------------------- | ------------------ | | Encode CloudEvents 1.0 structured JSON | `CloudEvent::encode`, `encode_cloud_event` | `encodeCloudEvent` | `encode_cloud_event` | `EncodeCloudEvent` | | Decode CloudEvents 1.0 structured JSON | `CloudEvent::decode`, `decode_cloud_event` | `decodeCloudEvent` | `decode_cloud_event` | `DecodeCloudEvent` | | Import explicit environment aliases | `import_env_config` | `importEnvConfig` | `import_env_config` | `ImportEnvConfig` | | Export canonical environment names | `export_env_config` | `exportEnvConfig` | `export_env_config` | `ExportEnvConfig` | ## Native JSON handles [Section titled “Native JSON handles”](#native-json-handles) Use raw methods for bytes and exact cross-language contracts. Use handles when the payload is app JSON and you want the name, codec, and type to stay together. * Node ```ts type SendEmail = { to: string; template: string }; const emails = forge.queue("emails"); await emails.enqueue({ to: "a@b.c", template: "welcome" }); const profile = forge.kv<{ name: string }>("user:42:profile"); await profile.set({ name: "Ada" }); const events = forge.topic<{ type: "created"; userId: string }>("user.created"); await events.publish({ type: "created", userId: "42" }); ``` * Python ```python emails = forge.queue("emails") await emails.enqueue({"to": "a@b.c", "template": "welcome"}) profile = forge.kv("user:42:profile") await profile.set({"name": "Ada"}) events = forge.topic("user.created") await events.publish({"type": "created", "user_id": "42"}) ``` * Rust ```rust use forgelib::{KvKey, KvTyped, QueuePayload, QueueTyped, QueueName, SetOpts}; #[derive(Serialize, Deserialize)] struct SendEmail { to: String, template: String } impl QueuePayload for SendEmail { const QUEUE: QueueName = QueueName::new("emails"); } let profile_key = KvKey::::new("user:42:profile"); forge.queue().enqueue_typed(&email).await?; forge.kv().set_typed(&profile_key, &profile, SetOpts::new()).await?; ``` ## Return types [Section titled “Return types”](#return-types) A few methods return small records. The field names match the table; Node uses camelCase and times end in `Ms`. * **Decision** (rate limit): `allowed`, `limit`, `remaining`, `resetAfterSeconds`, `retryAfterSeconds`. * **Job** (queue): `id`, `receipt`, `payload`, `attempt`, `maxAttempts`, plus the queue and lease deadline. * **QueueDepth**: `visible`, `inFlight`, `delayed`. * **BlobInfo** (head): `key`, `size`, `contentType`, `etag`, `lastModified`, `metadata`. * **BlobSummary** (list): `key`, `size`, opaque `etag`, `lastModified`; use head for content type and user metadata. * **ProxyPresign**: `url`, `method`, `key`, `expiresEpoch`, `maxBytes`, `signature`, `requiredHeaders`. * **NativePresign**: `url`, `method`, `expiresEpoch`, `requiredHeaders`, `constraints`; native PUT has no portable size ceiling. * **Session**: `userId`, `createdAt`, `expiresAt`. * **ApiKey** (on create): `id`, `secret` (shown once), `label`, `createdAt`, optional `expiresAt`, `scopes`, `metadata`. * **TokenConsumption**: `userId`, bounded opaque `payload`; returned atomically once. * **BackendInfo**: `primitive`, `provider`, `durable`, `caveats`. ## Errors [Section titled “Errors”](#errors) Every failure maps onto one of these. The retryable column says whether trying again might succeed. | Error | Retryable | Meaning | | -------------- | --------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `Config` | no | Bad configuration. Only happens during `init()`. | | `Unavailable` | yes | A transient backend outage (timeout, dropped connection). | | `NotFound` | no | The requested entity doesn’t exist. | | `Precondition` | no | A condition failed: a CAS mismatch, a lost lease, an unknown receipt. Re-read state and decide. (A duplicate `dedup_id` is not an error — enqueue returns the existing job’s id.) | | `Limit` | no | A size or quota limit was exceeded. | | `Invalid` | no | A bad argument or malformed input. | | `Backend` | sometimes | Anything else, carrying its own retryable flag. | In Rust these are variants of `ForgeError` with `code()`, `is_retryable()`, `operation()`, `backend_id()`, and a retained source chain. Node throws a structured `ForgeError` with `code`, `retryable`, `operation`, `backend`, `safeMessage`, and the normal `message`. Python raises typed subclasses of `forgelib.ForgeError` with the same fields. Go returns `*forge.Error` with the same stable metadata and an unwrapped local cause. Note Error messages are written so you can show or log them safely. The underlying cause, which might name a constraint or table, is kept on the error’s source and never put in the displayed text. ## Limits [Section titled “Limits”](#limits) Oversized input is rejected with a `Limit` error rather than truncated. | Thing | Limit | | --------------- | ------------- | | Key/value key | 512 bytes | | Key/value value | 1 MiB | | Queue payload | 256 KiB | | Blob object | 50 MiB | | Blob key | 1024 bytes | | Pub/sub payload | \~7 KB, UTF-8 | | Config value | 64 KiB | | Password input | 4096 bytes | | Token purpose | 255 bytes | ## Custom backends [Section titled “Custom backends”](#custom-backends) Forge ships memory and PostgreSQL for every primitive, plus filesystem and S3-compatible storage for blob bytes. Rust may inject a custom primitive trait at compile time as an advanced extension. Other languages intentionally have no plugin loader or arbitrary backend injection. ```rust let forge = forgelib::Forge::builder() .kv(my_redis_kv) // kv runs on your backend; the rest stay on Postgres .build() .await?; ``` ## Building with an agent [Section titled “Building with an agent”](#building-with-an-agent) Forge’s API isn’t in any model’s training data, so a coding agent won’t know it cold. Install the companion skill and the agent learns which primitive fits which task and the idioms for each language before it writes Forge code. ```sh npx skills add isala404/forge ``` ## More [Section titled “More”](#more) * [Examples](https://github.com/isala404/forge/tree/main/examples): one canonical application example per language, kept small and run in CI. * [Source and issues](https://github.com/isala404/forge) on GitHub. # Scope and costs > What Forge owns, what each hot path costs, and when to use a service directly. Forge is an in-process infrastructure library for application backends. It gives Rust, JavaScript, Python, and Go applications the same bounded primitives while PostgreSQL remains the durable system of record. It does not own application domain tables, HTTP routing, authorization policy, signals, deployment, or a hosted control plane. ## Ownership [Section titled “Ownership”](#ownership) | Forge owns | The application owns | | ------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------- | | Primitive APIs and the cross-language contract | HTTP, RPC, GraphQL, MCP, and other protocol surfaces | | PostgreSQL and memory behavior, plus filesystem and S3 blob adapters | Business tables, domain transactions, and authorization policy | | Forge schema, migrations, validation, and runtime diagnostics | Deployment order, process signals, readiness routes, and pool budgets | | Queue leases, retry state, dead-letter operations, and managed worker mechanics | Worker handlers, side-effect idempotency, cancellation checks, and audit records | | Bounded CloudEvents, trace-context, environment, queue-envelope, and invalidation encodings | Transport delivery, telemetry export, secret storage, and retention policy | | Server-side config caches and invalidation publication | Frontend framework, browser cache, TanStack Query, WebSocket or SSE client, reconnect, and refetch logic | Forge may provide a server-side building block without owning the surrounding system. For example, it can lease a queue job, but the application writes the handler. It can publish an invalidation hint, but the application authenticates a transport connection and the client refetches authoritative state. ## Partial adoption [Section titled “Partial adoption”](#partial-adoption) Use only the primitives the application needs. A normal PostgreSQL client still validates one coordinated Forge schema and opens one bounded system pool; unused primitives do no background work and have no feature-enable switches. A measured hot primitive can move to its own PostgreSQL target without changing its schema epoch or public contract. S3 may own large blob bytes while PostgreSQL continues to own Forge metadata and coordination. Memory mode is the explicit exception. It initializes without a database and keeps the same primitive APIs for tests and local development, but it is process-local, non-durable, and unsuitable for shared production coordination. ## Hot-path costs [Section titled “Hot-path costs”](#hot-path-costs) | Operation | Normal backend work | Important contention or size cost | | -------------------------- | --------------------------------------------------------------------- | ------------------------------------------------------------------------------------------ | | KV get/set/CAS | One indexed read or write | Large values consume database I/O; values are bounded at 1 MiB | | Queue enqueue | One indexed insert, plus an idempotency lookup when an ID is supplied | Payloads are bounded at 256 KiB; put larger artifacts in blob storage | | Queue dequeue | A bounded claim query using row locks and `SKIP LOCKED` | Polling and worker concurrency must fit the queue pool | | Pub/sub | PostgreSQL `NOTIFY` or an in-process broadcast | Connected-only and at-most-once; payload is about 7 KiB | | Rate limit | One atomic bucket update | Hot subjects serialize on the same bucket row; fail-open is an application risk decision | | Session/token verification | One indexed hash lookup | Password hashing is intentionally CPU-expensive; rate-limit before verifying passwords | | Schedule tick | One bounded due-item claim and queue insert per occurrence | Run ticks on every replica; deterministic IDs prevent duplicate occurrences | | PostgreSQL blob | Metadata plus byte I/O through PostgreSQL | Fine for bounded ordinary objects; use S3 for large or streaming bytes | | S3 blob | PostgreSQL metadata plus provider request(s) | Multipart, provider latency, egress, lifecycle, and bucket policy remain provider concerns | | Config/flags | Cache hit or one indexed read | Notifications accelerate invalidation; TTL refresh is the correctness fallback | Pool sizing is a capacity decision, not a language default to increase casually. Count API replicas, workers, schedulers, listeners, migration jobs, and feature-specific pools against PostgreSQL’s connection budget. Start with the conservative defaults, observe acquire latency and pool saturation, then change one measured bottleneck. ## Durability boundary [Section titled “Durability boundary”](#durability-boundary) An operation is durable only after its backend confirms success. PostgreSQL commits rows transactionally; queue delivery remains at-least-once because a worker can crash after performing application work and before acknowledging. Pub/sub is never durable. Filesystem blob bytes and PostgreSQL metadata are two writes. S3 owns byte durability after its successful response, while Forge owns the corresponding metadata. The [semantics table](/semantics/) states the exact boundary for every primitive. ## When direct use is better [Section titled “When direct use is better”](#when-direct-use-is-better) Use the underlying service or a dedicated product directly when its model is the product requirement rather than an implementation detail. Examples include Redis data structures Forge does not expose, Kafka-style durable ordered logs, SQS isolation at independent scale, provider-specific S3 lifecycle or legal-hold controls, OAuth/OIDC identity, a workflow engine, a synchronization engine, a billing ledger, vector search, or model routing. Do not force those requirements through KV, pub/sub, or queue merely to keep one dependency. Direct PostgreSQL tables are also the right answer for relational domain data, reporting joins, database constraints across business entities, and transactions owned by the application. Forge’s transactional outbox is the bridge when one of those domain transactions must reliably enqueue work. ## Supported backend admission [Section titled “Supported backend admission”](#supported-backend-admission) A backend is supported only when its capabilities and limitations are explicit, its generated contract stays aligned, and the applicable conformance, fault, upgrade, observability, package, and documentation checks pass. Forge does not emulate a guarantee the provider cannot make. PostgreSQL and memory are the supported runtime profiles; PostgreSQL, memory, filesystem, and S3-compatible storage are the supported blob roles described in [Primitive semantics](/semantics/). # Security guide > Namespace, credential, signed URL, rate-limit, and untrusted-payload guidance. Forge provides isolation mechanisms and bounded credential helpers. The application still authenticates requests, authorizes every resource action, owns HTTP security, and decides which failures may degrade safely. ## Namespace selection [Section titled “Namespace selection”](#namespace-selection) Choose one stable namespace per application and environment, such as `billing.production`. Never derive the global namespace from an untrusted request, and never reuse a production namespace in staging or tests against the same backend. Namespaces prevent accidental key and row collisions; they do not prove that the current user may access a tenant, queue, blob, or token. Authorize first, then compose a bounded primitive-specific key with the scoped-key helpers. ## Signed and presigned URLs [Section titled “Signed and presigned URLs”](#signed-and-presigned-urls) Treat every signed URL as a bearer credential. Send it only to the authorized caller, use short expiries, keep query strings out of logs and traces, and bind proxy validation to the expected method, namespace, logical key, format version, expiry, and upload limit. Forge proxy upload signatures are the portable choice when the application must enforce a hard byte ceiling or perform application authorization at upload time. Native S3 presigned GET/PUT avoids proxying bytes but cannot promise a portable maximum PUT body size; enforce bucket policy and validate the uploaded object before making it authoritative. Signing secrets need independent production values and normal secret-manager rotation. Do not place them in `forge.toml`, source control, metrics, exception messages, or diagnostics. Access keys, session tokens, database URLs, API-key secrets, session tokens, one-time tokens, and native presigned query strings follow the same rule. ## Sessions and cookies [Section titled “Sessions and cookies”](#sessions-and-cookies) Forge session tokens are opaque bearer credentials stored as hashes. When carried in cookies, the host framework must set `Secure`, `HttpOnly`, and an appropriate `SameSite` value, scope `Path` and `Domain` narrowly, and apply CSRF defenses to state-changing requests. Rotate the session after authentication or privilege changes, revoke it on logout, and use both idle and absolute expiry. Do not put session tokens in URLs or browser-readable telemetry. Password hashing uses Argon2id and `needsRehash` identifies records that should be upgraded after a successful verification. Rate-limit before password verification so credential stuffing cannot turn the password hasher into an unbounded CPU endpoint. Account-recovery responses should not reveal whether an identity exists. ## API keys and one-time tokens [Section titled “API keys and one-time tokens”](#api-keys-and-one-time-tokens) API-key secrets are returned once. Store only the displayed secret in the caller’s secret store; Forge stores its hash. Give keys a bounded expiry, explicit scopes, and small non-secret metadata, then enforce owner and scopes in application policy after verification. Revocation and expiry must be handled as normal unauthenticated results. One-time tokens are purpose-scoped, expiring, hashed, and atomically consumed. Keep payloads small and non-secret; use them for opaque return state or record IDs, not authorization policy. A token for one purpose cannot be replayed for another, and a consumed token cannot be recovered. ## Fail-open rate limits [Section titled “Fail-open rate limits”](#fail-open-rate-limits) Fail-open means a limiter backend outage allows the request. It is reasonable for low-risk availability paths where denial would cause a larger incident, but unsafe for password guessing, expensive compute, quota enforcement, abuse-sensitive writes, or financial limits. Select the failure mode per policy and route, alert on limiter backend failures, and never describe a fail-open limit as an enforced security boundary. ## Untrusted payloads [Section titled “Untrusted payloads”](#untrusted-payloads) KV values, queue bodies, pub/sub messages, blob metadata, config values, trace envelopes, and invalidation events are untrusted application input. Forge enforces byte and shape bounds, but the consumer must authenticate its source where relevant, validate its schema, avoid unsafe deserialization, and escape it for the output context. Queue handlers must be idempotent and must not execute payload strings as code, shell, SQL, templates, paths, or URLs without application validation. Use blob references for large queue artifacts and re-authorize access when resolving them. Default Forge telemetry excludes payloads, raw keys, identifiers, tokens, signed URLs, provider errors, and other high-cardinality content. Preserve that boundary in application spans and logs; opt-in model or user-content telemetry needs its own redaction and retention review. # Primitive semantics > Consistency, ordering, duplication, expiry, retry, outage, and multi-process guarantees for every primitive and backend. Forge keeps method shapes aligned across backends, but it does not pretend the backends have identical durability or process scope. This page is the behavioral contract; backend caveats returned by `backendCapabilities` are the runtime view of the same facts. ## Contract by primitive [Section titled “Contract by primitive”](#contract-by-primitive) | Primitive | Consistency and atomicity | Ordering and duplication | Expiry and retry | Outage and multi-process behavior | | ---------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | KV | A single get observes committed state. Set-if-absent, set-if-present, increment, expiry changes, and compare-and-swap are atomic per key. Prefix scans are ordered pages, not snapshots. | No event ordering is implied. A repeated unconditional write is last-commit-wins; CAS is the write fence. | TTL is checked on access and reclaimed by maintenance; an expired key is absent even before physical cleanup. Retry only retryable errors and use CAS or an idempotent value when duplicate writes matter. | PostgreSQL is durable and shared. Memory is process-local and lost on exit. An outage returns a retryable unavailable/backend error; Forge never silently falls back to memory. | | Queue | Enqueue commits before returning. Dequeue grants a time-bounded fenced lease; only its current receipt may heartbeat or settle. | Visible jobs are FIFO where priorities are equal, but concurrency, retries, and delayed jobs prevent a global completion order. Delivery is at-least-once. A caller job ID makes enqueue idempotent; a deduplication ID is a separate bounded reservation. | Delay controls first visibility. Nack or lease expiry schedules retry until `maxAttempts`, then the job becomes dead. Payload and terminal retention are separate settings. | PostgreSQL workers share claims across replicas. Memory workers coordinate only inside one process and lose jobs on exit. Backend outages do not turn a handled job into an acknowledgement; stale workers are fenced. | | Pub/sub | Publish reports provider acceptance, not subscriber processing. A subscriber sees only messages delivered while connected. | At-most-once, no replay, and no durable ordering promise across reconnects or publishers. | No retention, retry, cursor, or catch-up exists. Use a queue or application table when loss matters. | PostgreSQL uses cross-process `LISTEN`/`NOTIFY`; memory is in-process only. Reconnects may miss messages, so notifications must trigger a read of authoritative state. | | Blob | Put replaces one logical key subject to its precondition. Head returns full metadata; list returns summaries. PostgreSQL bytes are transactional with Forge metadata, while filesystem and S3 bytes are separate provider writes. | List is key-ordered pagination but not a snapshot under concurrent writes. Delete is idempotent and makes no prior-existence claim. ETags are opaque version tokens, not portable MD5 hashes. | Native provider retries are bounded. Buffered bodies are size-limited; range and streaming paths avoid whole-object buffering. Presigns expire at their declared instant and are bearer credentials. | PostgreSQL and S3 are shared and durable within their provider guarantees. Filesystem requires a shared mount across replicas. Memory is local and ephemeral. Provider outages return retryable errors and do not fall back to another byte store. | | Auth | Password verification is local CPU work. Session, API-key, and one-time-token records are stored as hashes; token consumption is atomic and single-winner. | Tokens have no ordering. Repeating create operations mints new credentials; repeating one-time-token consumption returns absent. | Idle and absolute session expiry, API-key expiry, and token expiry are checked on access and reclaimed later. Credential creation is not safe to blindly retry unless the application can discard all returned secrets. | PostgreSQL credentials work across replicas. Memory credentials exist in one process and vanish on exit. Backend unavailability is an authentication dependency failure; the application decides whether to reject or degrade. | | Rate limit | Check-and-consume is atomic per policy and subject. A denial is a decision, not an error. | Concurrent checks serialize at the bucket and cannot oversubscribe capacity. No ordering is exposed to callers. | Capacity refills from elapsed time. Retry after the returned interval; do not retry an allowed decision as though it were idempotent. | PostgreSQL enforces one shared budget. Memory creates one independent budget per process. Fail-open converts a backend outage into an allowed decision and must be chosen explicitly for the route’s risk. | | Schedule | Each selected occurrence becomes one deterministic queue job. Concurrent ticks claim rows, while the occurrence ID makes retries idempotent. | Cron and absolute instants are stored in UTC. `skip`, `run_once`, and capped `catch_up` define outage ordering; queue handling may still reorder completion. | One-offs fire at most once. Failed enqueue remains due. Catch-up is capped at 100 per schedule and drops older excess occurrences before advancing, preventing restart storms. | PostgreSQL schedules and scheduler diagnostics are durable and shared. Memory state is process-local. Pause removes a schedule from due lag/count without deleting it. | | Config and flags | Writes commit before cache invalidation is announced. Reads may use a bounded local cache; TTL refresh is the correctness fallback after missed notifications. | There is no event log. Concurrent writes are last-commit-wins. Flag evaluation is deterministic for the same key, subject, value, and captured environment override. | Cache entries expire by TTL. Notification reconnect uses bounded backoff. Environment overrides are captured at process initialization and do not change afterward. | PostgreSQL values are shared and invalidated across processes. Memory values and notifications are local. During outage a cached value may remain available until its documented stale bound; uncached reads fail. | Invalidation events use pub/sub only as a lossy transport hint. Their optional `revision` is application-owned metadata, not a Forge replay cursor. After connecting or reconnecting, a consumer reads authoritative application state or resumes from an application-owned durable cursor. ## Backend capabilities and limitations [Section titled “Backend capabilities and limitations”](#backend-capabilities-and-limitations) | Backend role | Durable | Shared across processes | Notifications | Streaming/range | Native presign | Primary limitations | | ------------------------ | ---------------------------------- | ------------------------ | ------------------------------------- | ------------------------------ | ---------------------- | ------------------------------------------------------------------------------------------------------------- | | PostgreSQL runtime | Yes after commit | Yes | Yes, connected-only | Blob range/streaming supported | No provider-native URL | Pool and row contention must be measured; pub/sub remains non-durable | | Memory profile | No | No | In-process only | Bounded in-memory bodies | No | Tests/local development only; restart loses all state; rate limits multiply per process | | Filesystem blob bytes | Byte files survive process restart | Only with a shared mount | Metadata notifications use PostgreSQL | Yes | No | Byte write and PostgreSQL metadata write are not one transaction | | S3-compatible blob bytes | Provider-defined after success | Yes | Metadata coordination uses PostgreSQL | Yes, including multipart | Yes | Provider policies, lifecycle, replication, legal hold, CDN, and native PUT size enforcement are outside Forge | ## Crash boundaries [Section titled “Crash boundaries”](#crash-boundaries) A successful queue acknowledgement means Forge committed settlement, not that an external side effect happened exactly once. Handlers must be idempotent. When an application-owned SQL transaction and enqueue must agree, insert the versioned outbox row in that transaction and let the relay enqueue by deterministic event ID. A crash before enqueue leaves the row pending; a crash after enqueue repeats the same effective job; a crash after marking is complete. Expiry is logical before it is physical. Reads and claims exclude expired state immediately according to backend time, while `maintain` later removes storage. Production replicas therefore need reasonably synchronized clocks; Forge diagnostics report meaningful skew where the provider exposes it. # Support matrix > Supported languages, runtimes, platforms, PostgreSQL versions, and package formats. Forge 1.1 has four first-party language surfaces. Node.js and Bun use the same JavaScript package and contract; Bun is a tested runtime, not a separate binding. | Language | Runtime floor | CI versions | Package | | ---------- | --------------------- | ------------------- | ----------------------------- | | Rust | 1.94.1 | 1.94.1 and stable | crates.io crate | | JavaScript | Node.js 24 or Bun 1.4 | Node.js 24; Bun 1.4 | npm package with native addon | | Python | 3.9 | 3.9 through 3.14 | wheel and sdist on PyPI | | Go | 1.27 | 1.27 on Linux amd64 | Go module | | Surface | Supported matrix | | ------------------ | ---------------------------------------------------------- | | PostgreSQL | 18 | | Linux | amd64 glibc, amd64 musl for Node, arm64 glibc | | macOS | arm64 | | Windows | amd64 | | Rust package | source crate | | JavaScript package | npm tarball plus platform native addon | | Python package | abi3 wheels plus sdist | | Go package | pure-Go source module; tested and supported on Linux amd64 | The declared runtime and PostgreSQL versions are release gates. A platform listed here is supported only when its packaged artifact installs and its smoke test passes. # Build a chat app (Node) > A complete Node and Hono REST API for a realtime group chat with presence, typing, attachments, and disappearing messages, on six Forge primitives. You’ll build a realtime group chat in TypeScript with [Hono](https://hono.dev): messages arrive live over server-sent events, people see who’s online and who’s typing, files upload straight to storage, and messages can disappear on a timer. Pub/sub carries everything live, a queue fans each message out to update unread counts, key/value stores messages and presence and de-duplicates sends, blob holds attachments, schedule runs the disappearing timer, and rate limiting guards sending and uploading. This is the Node version, built as a REST and SSE API. The example app in the repo is the [chatapp](https://github.com/isala404/forge/tree/main/examples/chatapp), which does the same thing over GraphQL subscriptions. ## Start the project [Section titled “Start the project”](#start-the-project) package.json ```json { "name": "chatapp", "type": "module", "dependencies": { "@hono/node-server": "^1", "hono": "^4", "forgelib": "^1" } } ``` The presigned upload URLs need a signing secret, so set one in `forge.toml`. forge.toml ```toml [postgres] url = "${DATABASE_URL:-postgres://postgres:forge@localhost:5432/chatapp}" [blob] signing_secret = "${BLOB_SIGNING_SECRET:-dev-secret-change-me}" ``` The whole app is two files plus config: * src * server.ts init Forge, the routes, start the workers * worker.ts the background workers * package.json * forge.toml ## The server [Section titled “The server”](#the-server) Initialize Forge once, start a worker and the scheduler loop, and wire up Hono. A top-level `await` works because this is an ES module. The handlers below all close over this `forge` handle. src/server.ts ```ts import { randomUUID } from "node:crypto"; import { serve } from "@hono/node-server"; import { Hono } from "hono"; import { cors } from "hono/cors"; import { streamSSE } from "hono/streaming"; import { ForgeClient } from "forgelib"; import { runFanoutWorker, runReapWorker, runScheduler } from "./worker.ts"; const forge = await ForgeClient.init(); // reads ./forge.toml, runs migrations runFanoutWorker(forge); runReapWorker(forge); runScheduler(forge); const app = new Hono(); app.use("*", cors()); function bearer(c: { req: { header: (k: string) => string | undefined } }): string { return (c.req.header("authorization") ?? "").replace(/^Bearer /i, "").trim(); } async function requireUser(c: any): Promise { return forge.validateSession(bearer(c)); } // ... routes go here ... serve({ fetch: app.fetch, port: 8787 }); console.log("chat on http://localhost:8787"); ``` ## Accounts [Section titled “Accounts”](#accounts) Sign-up and login are the same shape as the other tutorials: throttle by email, hash the password, claim the email with a “set if not present” write, and open a session. src/server.ts (continued) ```ts app.post("/api/signup", async (c) => { const { email, password } = await c.req.json(); const limit = await forge.rateLimitCheck("chat-auth", email, 20, 60, true); if (!limit.allowed) return c.json({ error: "slow down" }, 429); const user = { id: randomUUID(), email, passwordHash: await forge.hashPassword(password) }; const created = await forge.kvSet(`user:email:${email}`, JSON.stringify(user), null, true); if (!created) return c.json({ error: "email already registered" }, 409); await forge.kvSet(`user:id:${user.id}`, JSON.stringify(user)); const token = await forge.createSession(user.id); return c.json({ token, userId: user.id }, 201); }); app.post("/api/login", async (c) => { const { email, password } = await c.req.json(); const limit = await forge.rateLimitCheck("chat-auth", email, 20, 60, true); if (!limit.allowed) return c.json({ error: "slow down" }, 429); const raw = await forge.kvGet(`user:email:${email}`); if (!raw) return c.json({ error: "invalid email or password" }, 401); const user = JSON.parse(raw); if (!(await forge.verifyPassword(password, user.passwordHash))) return c.json({ error: "invalid email or password" }, 401); return c.json({ token: await forge.createSession(user.id), userId: user.id }); }); ``` ## Send a message [Section titled “Send a message”](#send-a-message) Sending throttles the sender (fail open, so a limiter blip never blocks chat), de-duplicates with an idempotency key so a retried request can’t post twice, stores the message, publishes it live to the room, and queues the fan-out work. If the room has disappearing messages on, it also schedules the message’s deletion. src/server.ts (continued) ```ts const roomMsgsKey = (room: string) => `room:${room}:msgs`; app.post("/api/rooms/:room/messages", async (c) => { const userId = await requireUser(c); if (!userId) return c.json({ error: "authentication required" }, 401); const room = c.req.param("room"); const { body, idempotencyKey, mediaKey } = await c.req.json(); const d = await forge.rateLimitCheck("send", userId, 30, 10, true); if (!d.allowed) return c.json({ error: "you are sending too fast" }, 429); const id = randomUUID(); if (idempotencyKey) { const won = await forge.kvSet(`idem:${userId}:${idempotencyKey}`, id, 86400, true); if (!won) return c.json({ error: "duplicate send" }, 409); } const msg = { id, room, userId, body, mediaKey: mediaKey ?? null, at: Date.now() }; // keep the last 50 messages per room in key/value const recent = JSON.parse((await forge.kvGet(roomMsgsKey(room))) ?? "[]"); recent.push(msg); await forge.kvSet(roomMsgsKey(room), JSON.stringify(recent.slice(-50))); // live now, fan-out later; dedup the queue job on the message id await forge.topic<{ type: "message"; message: typeof msg }>(`chat:${room}`) .publish({ type: "message", message: msg }); await forge.queue<{ room: string; id: string }>("fanout").enqueue({ room, id }, { dedupId: id }); // disappearing messages: schedule a delete if the room has a TTL set const ttl = await forge.kvGet(`room:${room}:ttl`); if (ttl) await forge.scheduleAt(Date.now() + Number(ttl) * 1000, "reap", JSON.stringify({ room, id })); return c.json(msg, 201); }); app.get("/api/rooms/:room/messages", async (c) => { if (!(await requireUser(c))) return c.json({ error: "authentication required" }, 401); const recent = JSON.parse((await forge.kvGet(roomMsgsKey(c.req.param("room")))) ?? "[]"); return c.json({ messages: recent }); }); ``` ## Stream a room [Section titled “Stream a room”](#stream-a-room) The client opens an SSE stream per room. The route subscribes to the room topic and forwards each event (a new message, a typing flag) until the client disconnects, then drops the subscription. src/server.ts (continued) ```ts app.get("/api/rooms/:room/stream", async (c) => { if (!(await requireUser(c))) return c.json({ error: "authentication required" }, 401); const room = c.req.param("room"); return streamSSE(c, async (stream) => { const events = await forge.topic(`chat:${room}`).subscribe(); stream.onAbort(() => { void events.return?.(); }); for await (const event of events) { await stream.writeSSE({ data: JSON.stringify(event) }); } }); }); ``` ## Typing and presence [Section titled “Typing and presence”](#typing-and-presence) Typing is purely live, so it’s a publish with nothing stored. Presence is a key/value key with a short TTL that a heartbeat refreshes: someone is online while their key is alive, and it expires on its own when the heartbeats stop. src/server.ts (continued) ```ts app.post("/api/rooms/:room/typing", async (c) => { const userId = await requireUser(c); if (!userId) return c.json({ error: "authentication required" }, 401); await forge.topic<{ type: "typing"; userId: string }>(`chat:${c.req.param("room")}`) .publish({ type: "typing", userId }); return c.body(null, 204); }); app.post("/api/heartbeat", async (c) => { const userId = await requireUser(c); if (!userId) return c.json({ error: "authentication required" }, 401); await forge.kvSet(`presence:${userId}`, "1", 30); // expires in 30s await forge.topic<{ userId: string; online: boolean }>("presence").publish({ userId, online: true }); return c.body(null, 204); }); app.get("/api/presence/:userId", async (c) => { const online = await forge.kvExists(`presence:${c.req.param("userId")}`); return c.json({ online }); }); ``` ## Attach a file [Section titled “Attach a file”](#attach-a-file) Uploads go straight to storage, not through your server. The client asks for a presigned upload URL, PUTs the file to it, then sends a message that references the key. Throttle this one fail-closed, since minting upload URLs is more expensive than a normal request, and read the size cap from config so you can change it without a deploy. src/server.ts (continued) ```ts app.post("/api/rooms/:room/upload-url", async (c) => { const userId = await requireUser(c); if (!userId) return c.json({ error: "authentication required" }, 401); const room = c.req.param("room"); const rl = await forge.rateLimitCheck("upload", userId, 10, 60, false); // fail closed if (!rl.allowed) return c.json({ error: "too many uploads" }, 429); const maxBytes = Number(await forge.configGet("max_upload_bytes")) || 10 * 1024 * 1024; const key = `media/${room}/${randomUUID()}`; const uploadUrl = (await forge.blobPresignUpload(key, 600, maxBytes)).url; return c.json({ key, uploadUrl, maxBytes }); }); ``` ## The workers [Section titled “The workers”](#the-workers) Two background jobs finish the picture. The `fanout` worker takes each sent message and bumps an unread counter for every room member except the sender. The `reap` worker, fed by the disappearing-message schedules, removes a message when its timer fires. The scheduler loop ticks the timers and runs the maintenance sweep. src/worker.ts ```ts import type { ForgeClient } from "forgelib"; const roomMsgsKey = (room: string) => `room:${room}:msgs`; export function runFanoutWorker(forge: ForgeClient): void { void forge.worker<{ room: string; id: string }>("fanout", async (job) => { const { room, id } = job.payload; const recent = JSON.parse((await forge.kvGet(roomMsgsKey(room))) ?? "[]"); const msg = recent.find((m: any) => m.id === id); const members = JSON.parse((await forge.kvGet(`room:${room}:members`)) ?? "[]"); for (const member of members) { if (member !== msg?.userId) await forge.kvIncr(`unread:${member}:${room}`, 1); } }); } export function runReapWorker(forge: ForgeClient): void { void forge.worker<{ room: string; id: string }>("reap", async (job) => { const { room, id } = job.payload; const recent = JSON.parse((await forge.kvGet(roomMsgsKey(room))) ?? "[]"); await forge.kvSet(roomMsgsKey(room), JSON.stringify(recent.filter((m: any) => m.id !== id))); await forge.topic<{ type: "deleted"; id: string }>(`chat:${room}`).publish({ type: "deleted", id }); }); } export function runScheduler(forge: ForgeClient): void { void (async () => { for (;;) { try { await forge.runSchedulerOnce(); await forge.maintain(); } catch (e) { console.warn("scheduler tick failed:", e); } await new Promise((r) => setTimeout(r, 30_000)); } })(); } ``` The reap worker publishes a `deleted` event too, so open clients drop the message from view the moment it disappears. ## Run it [Section titled “Run it”](#run-it) ```sh createdb chatapp npm install node src/server.ts ``` Terminal ```sh TOKEN=$(curl -s localhost:8787/api/signup -H 'content-type: application/json' \ -d '{"email":"a@b.com","password":"supersecret"}' | jq -r .token) # open a live stream in one terminal curl -N localhost:8787/api/rooms/general/stream -H "authorization: Bearer $TOKEN" # send a message in another and watch it arrive on the stream curl -s localhost:8787/api/rooms/general/messages \ -H "authorization: Bearer $TOKEN" -H 'content-type: application/json' \ -d '{"body":"hello"}' ``` That’s a realtime app where the live layer, the durable layer, and the background work all run on one Postgres connection. From here, the [Primitives](/primitives/) page has the details behind each call, and [Operations](/operations/) covers running the workers and scheduler in production. [All the primitives](/primitives/)The methods and gotchas behind everything you just used. # Build a to-do app (Rust) > A complete Rust and Rocket REST API for a multi-user to-do app, built end to end on Forge's auth, key/value, rate limit, and queue primitives. You’ll build a working REST API in Rust with [Rocket](https://rocket.rs): people sign up, log in, and keep their own to-do list, and every change drops an event on a background audit queue that a worker drains. It’s the smallest app that still uses four primitives, so it’s a good first build. It uses **auth** for password hashing and sessions, **key/value** for user records and each person’s list, **rate limit** to throttle sign-up and login, and **queue** for the audit trail. This tutorial follows the canonical Rust [todo app](https://github.com/isala404/forge/tree/main/examples/todoapp). The examples directory also has one small application for each other supported language. ## Start the project [Section titled “Start the project”](#start-the-project) Create a binary crate and add the dependencies. Forge is `forgelib`; the rest is a normal Rocket setup. Cargo.toml ```toml [package] name = "todoapp" version = "0.1.0" edition = "2024" [dependencies] forgelib = "1" rocket = { version = "0.5", features = ["json"] } serde = { version = "1", features = ["derive"] } serde_json = "1" tokio = { version = "1", features = ["rt-multi-thread", "macros", "signal", "time"] } uuid = { version = "1", features = ["v4"] } chrono = { version = "0.4", features = ["serde"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } anyhow = "1" ``` Forge reads its config from a `forge.toml` next to the crate. One connection string is enough; everything else takes a default. forge.toml ```toml [postgres] url = "${DATABASE_URL:-postgres://postgres:forge@localhost:5432/todoapp}" ``` The code lives in a few small modules: * src * main.rs init Forge, mount routes, spawn the worker * routes.rs the handlers * types.rs request/response types and the error type * util.rs validation and key helpers * Cargo.toml * forge.toml ## The error type [Section titled “The error type”](#the-error-type) Handlers return `Result`. The trick that keeps them readable is teaching `ApiError` to convert from a `ForgeError`, so a failed Forge call propagates with `?` and turns into a 500 with a safe message. Validation paths build an `ApiError` directly with the right status. src/types.rs ```rust use rocket::http::Status; use rocket::response::{status, Responder}; use rocket::serde::json::Json; use rocket::Request; use serde::{Deserialize, Serialize}; #[derive(Serialize)] pub struct ErrorBody { pub error: String } #[derive(Debug)] pub struct ApiError { pub status: Status, pub message: String } impl ApiError { pub fn new(status: Status, msg: impl Into) -> Self { Self { status, message: msg.into() } } pub fn bad_request(m: impl Into) -> Self { Self::new(Status::BadRequest, m) } pub fn unauthorized() -> Self { Self::new(Status::Unauthorized, "authentication required") } pub fn conflict(m: impl Into) -> Self { Self::new(Status::Conflict, m) } pub fn not_found(m: impl Into) -> Self { Self::new(Status::NotFound, m) } } impl<'r> Responder<'r, 'static> for ApiError { fn respond_to(self, req: &'r Request<'_>) -> rocket::response::Result<'static> { status::Custom(self.status, Json(ErrorBody { error: self.message })).respond_to(req) } } // `?` on a Forge call now yields a 500 without leaking the cause. impl From for ApiError { fn from(err: forgelib::ForgeError) -> Self { tracing::warn!(error = %err, "forge call failed"); Self::new(Status::InternalServerError, "internal error") } } impl From for ApiError { fn from(_: serde_json::Error) -> Self { Self::new(Status::InternalServerError, "internal error") } } ``` The data types are plain structs. The list is stored as one JSON array per user. src/types.rs (continued) ```rust #[derive(Serialize, Deserialize, Clone)] pub struct UserRecord { pub id: String, pub email: String, pub password_hash: String } #[derive(Deserialize)] pub struct Credentials { pub email: String, pub password: String } #[derive(Serialize)] pub struct AuthResponse { pub token: String, pub email: String } #[derive(Serialize, Deserialize, Clone)] pub struct Todo { pub id: String, pub title: String, pub completed: bool, } #[derive(Deserialize)] pub struct TodoCreate { pub title: String } #[derive(Deserialize)] pub struct TodoPatch { pub title: Option, pub completed: Option } ``` ## Helpers [Section titled “Helpers”](#helpers) A handful of small functions: validate input, pull the bearer token off the header, and build the key/value keys. Namespacing the keys (`todo:...`) keeps them tidy if this database is ever shared. src/util.rs ```rust use crate::types::{ApiError, Credentials}; pub fn validate_credentials(c: &Credentials) -> Result<(String, String), ApiError> { let email = c.email.trim().to_ascii_lowercase(); let password = c.password.trim().to_string(); if !email.contains('@') || email.len() > 254 { return Err(ApiError::bad_request("enter a valid email")); } if password.len() < 8 { return Err(ApiError::bad_request("password must be at least 8 characters")); } Ok((email, password)) } pub fn validate_title(raw: &str) -> Result { let title = raw.trim(); if title.is_empty() || title.len() > 160 { return Err(ApiError::bad_request("title must be 1 to 160 characters")); } Ok(title.to_string()) } pub fn bearer_token(raw: Option<&str>) -> Result { let raw = raw.ok_or_else(ApiError::unauthorized)?; let token = raw.strip_prefix("Bearer ").unwrap_or(raw).trim(); if token.is_empty() { return Err(ApiError::unauthorized()); } Ok(token.to_string()) } pub fn user_email_key(email: &str) -> String { format!("todo:user:email:{email}") } pub fn user_id_key(id: &str) -> String { format!("todo:user:id:{id}") } pub fn todos_key(uid: &str) -> String { format!("todo:list:{uid}") } ``` ## Sign up and log in [Section titled “Sign up and log in”](#sign-up-and-log-in) Sign-up throttles by email first, hashes the password, then writes the user under its email key with `SetMode::IfNotExists`. That write is the uniqueness check: it returns `false` if the email is taken, so there’s no read-then-write race. Then it opens a session and returns the token. src/routes.rs ```rust use std::time::Duration; use forgelib::{Bytes, FailMode, Forge, Limit, PhcString, SessionOpts, SetMode, SetOpts}; use rocket::http::{Header, Status}; use rocket::request::{FromRequest, Outcome}; use rocket::response::status; use rocket::serde::json::Json; use rocket::{delete, get, patch, post, Request, State}; use uuid::Uuid; use crate::types::*; use crate::util::*; pub struct AppState { pub forge: Forge } const AUDIT_QUEUE: &str = "todo-audit"; #[post("/api/signup", data = "")] async fn signup(state: &State, input: Json) -> Result>, ApiError> { let (email, password) = validate_credentials(&input)?; let limit = state.forge.ratelimit() .check_with("todo-auth", &email, Limit::per_duration(20, Duration::from_secs(60)), FailMode::Open).await?; if !limit.allowed { return Err(ApiError::new(Status::TooManyRequests, "too many attempts; slow down")); } let user = UserRecord { id: Uuid::new_v4().to_string(), email: email.clone(), password_hash: state.forge.auth().hash_password(&password).await?.as_str().to_string(), }; let body = Bytes::from(serde_json::to_vec(&user)?); let created = state.forge.kv() .set(&user_email_key(&email), body.clone(), SetOpts::new().with_mode(SetMode::IfNotExists)) .await?; if !created { return Err(ApiError::conflict("email already registered")); } state.forge.kv().set(&user_id_key(&user.id), body, SetOpts::new()).await?; let token = state.forge.auth() .create_session(&user.id, SessionOpts::new() .with_idle_timeout(Duration::from_secs(30 * 60)) .with_absolute_timeout(Duration::from_secs(7 * 24 * 60 * 60))) .await?; Ok(status::Custom(Status::Created, Json(AuthResponse { token: token.as_str().to_string(), email }))) } ``` Login is the same throttle, then read the user, verify the password against the stored hash, and start a session. src/routes.rs (continued) ```rust #[post("/api/login", data = "")] async fn login(state: &State, input: Json) -> Result, ApiError> { let (email, password) = validate_credentials(&input)?; let limit = state.forge.ratelimit() .check_with("todo-auth", &email, Limit::per_duration(20, Duration::from_secs(60)), FailMode::Open).await?; if !limit.allowed { return Err(ApiError::new(Status::TooManyRequests, "too many attempts; slow down")); } let Some(raw) = state.forge.kv().get(&user_email_key(&email)).await? else { return Err(ApiError::new(Status::Unauthorized, "invalid email or password")); }; let user: UserRecord = serde_json::from_slice(&raw)?; let ok = state.forge.auth() .verify_password(&password, &PhcString::new(user.password_hash.clone())).await?; if !ok { return Err(ApiError::new(Status::Unauthorized, "invalid email or password")); } let token = state.forge.auth() .create_session(&user.id, SessionOpts::default()).await?; Ok(Json(AuthResponse { token: token.as_str().to_string(), email: user.email })) } ``` ## A request guard for auth [Section titled “A request guard for auth”](#a-request-guard-for-auth) Every list route needs the same thing: read the `Authorization` header, validate the session, and get the user id. A small Rocket request guard reads the header, and a helper turns it into a user id or a 401. src/routes.rs (continued) ```rust pub struct AuthHeader(Option); #[rocket::async_trait] impl<'r> FromRequest<'r> for AuthHeader { type Error = std::convert::Infallible; async fn from_request(req: &'r Request<'_>) -> Outcome { Outcome::Success(Self(req.headers().get_one("Authorization").map(str::to_string))) } } async fn require_user(state: &AppState, auth: &AuthHeader) -> Result { let token = bearer_token(auth.0.as_deref())?; match state.forge.auth().validate_session(&token).await? { Some(session) => Ok(session.user_id), None => Err(ApiError::unauthorized()), } } ``` ## The to-do routes [Section titled “The to-do routes”](#the-to-do-routes) The list lives under one key as a JSON array. Each route reads it, changes it, and writes it back, then enqueues an audit event. The `dedup_id` is the action plus the todo id, so a double-submit collapses to one audit record, and `max_attempts(3)` caps retries. src/routes.rs (continued) ```rust async fn load_todos(state: &AppState, uid: &str) -> Result, ApiError> { Ok(match state.forge.kv().get(&todos_key(uid)).await? { Some(bytes) => serde_json::from_slice(&bytes)?, None => Vec::new(), }) } async fn save_todos(state: &AppState, uid: &str, todos: &[Todo]) -> Result<(), ApiError> { state.forge.kv() .set(&todos_key(uid), Bytes::from(serde_json::to_vec(todos)?), SetOpts::new()).await?; Ok(()) } async fn audit(state: &AppState, uid: &str, action: &str, todo_id: &str) -> Result<(), ApiError> { use forgelib::EnqueueOpts; let event = serde_json::json!({ "userId": uid, "action": action, "todoId": todo_id }); state.forge.queue() .enqueue(AUDIT_QUEUE, Bytes::from(event.to_string()), EnqueueOpts::new().with_max_attempts(3).with_dedup_id(format!("{action}:{todo_id}"))) .await?; Ok(()) } #[get("/api/todos")] async fn list_todos(state: &State, auth: AuthHeader) -> Result>, ApiError> { let uid = require_user(state, &auth).await?; Ok(Json(load_todos(state, &uid).await?)) } #[post("/api/todos", data = "")] async fn create_todo(state: &State, auth: AuthHeader, input: Json) -> Result>, ApiError> { let uid = require_user(state, &auth).await?; let mut todos = load_todos(state, &uid).await?; let todo = Todo { id: Uuid::new_v4().to_string(), title: validate_title(&input.title)?, completed: false }; todos.insert(0, todo.clone()); save_todos(state, &uid, &todos).await?; audit(state, &uid, "created", &todo.id).await?; Ok(status::Custom(Status::Created, Json(todo))) } #[patch("/api/todos/", data = "")] async fn update_todo(state: &State, auth: AuthHeader, id: &str, input: Json) -> Result, ApiError> { let uid = require_user(state, &auth).await?; let mut todos = load_todos(state, &uid).await?; let todo = todos.iter_mut().find(|t| t.id == id) .ok_or_else(|| ApiError::not_found("todo not found"))?; if let Some(title) = &input.title { todo.title = validate_title(title)?; } if let Some(done) = input.completed { todo.completed = done; } let updated = todo.clone(); save_todos(state, &uid, &todos).await?; audit(state, &uid, "updated", &updated.id).await?; Ok(Json(updated)) } #[delete("/api/todos/")] async fn delete_todo(state: &State, auth: AuthHeader, id: &str) -> Result { let uid = require_user(state, &auth).await?; let mut todos = load_todos(state, &uid).await?; let before = todos.len(); todos.retain(|t| t.id != id); if todos.len() == before { return Err(ApiError::not_found("todo not found")); } save_todos(state, &uid, &todos).await?; audit(state, &uid, "deleted", id).await?; Ok(Status::NoContent) } ``` ## Drain the audit queue [Section titled “Drain the audit queue”](#drain-the-audit-queue) The handlers only enqueue. Something has to drain the queue, and Forge doesn’t run that loop for you. Rust ships a managed worker that leases jobs, runs your handler, acks on `Ok` and nacks on `Err`, and shuts down cleanly. Here it just logs each event, but this is where you’d write to an audit table or ship it off. src/worker.rs ```rust use forgelib::{Forge, Job}; pub async fn run_audit_worker(forge: Forge) { forge.worker("todo-audit") .run(|job: Job| async move { let event: serde_json::Value = job.payload_json()?; tracing::info!(?event, "audit"); Ok::<(), serde_json::Error>(()) }) .await; } ``` ## Wire it up and run [Section titled “Wire it up and run”](#wire-it-up-and-run) `main.rs` reads `forge.toml`, spawns the worker and a maintenance loop, and launches Rocket with the `Forge` handle in shared state. src/main.rs ```rust mod routes; mod types; mod util; mod worker; use std::time::Duration; use forgelib::Forge; use routes::AppState; #[rocket::main] async fn main() -> anyhow::Result<()> { tracing_subscriber::fmt().with_env_filter("info,sqlx=warn").init(); let forge = Forge::init().await?; // reads ./forge.toml, runs migrations tokio::spawn(worker::run_audit_worker(forge.clone())); tokio::spawn({ let forge = forge.clone(); async move { loop { let _ = forge.run_scheduler_once().await; let _ = forge.maintain().await; tokio::time::sleep(Duration::from_secs(30)).await; } } }); rocket::build() .manage(AppState { forge }) .mount("/", rocket::routes![ routes::signup, routes::login, routes::list_todos, routes::create_todo, routes::update_todo, routes::delete_todo, ]) .launch() .await?; Ok(()) } ``` Start Postgres, then run it: ```sh createdb todoapp cargo run ``` And exercise it: Terminal ```sh # sign up, capture the token TOKEN=$(curl -s localhost:8000/api/signup \ -H 'content-type: application/json' \ -d '{"email":"a@b.com","password":"supersecret"}' | jq -r .token) # add a todo curl -s localhost:8000/api/todos \ -H "authorization: Bearer $TOKEN" -H 'content-type: application/json' \ -d '{"title":"write the docs"}' # list them curl -s localhost:8000/api/todos -H "authorization: Bearer $TOKEN" ``` That’s the whole app: accounts, a per-user list, throttled auth, and a drained audit queue, all on one Postgres connection. The URL shortener pulls in the other four primitives, in Python. [Build a URL shortener (Python)](/tutorials/url-shortener/)Blob, pub/sub, schedule, and config, end to end in FastAPI. # Build a URL shortener (Python) > A complete Python and FastAPI REST API for a link shortener with click analytics, QR codes, expiry, and live counts, touching all eight Forge primitives. You’ll build a link shortener in Python with [FastAPI](https://fastapi.tiangolo.com): people create short links, every redirect bumps a counter, a dashboard updates live, each link carries a QR image, links can expire, and a feature flag gates custom slugs. This one touches all eight primitives. The idea that ties it together is the redirect path. It does the cheap work inline (bump the counter, check a rate limit) and hands everything else to a queue, so the redirect stays fast and a worker fans out the rest. This tutorial follows the canonical Python [link shortener](https://github.com/isala404/forge/tree/main/examples/linksapp). The examples directory also has one small application for each other supported language. ## Start the project [Section titled “Start the project”](#start-the-project) pyproject.toml ```toml [project] name = "linksapp" version = "0.1.0" requires-python = ">=3.12" dependencies = [ "fastapi>=0.115", "uvicorn>=0.34", "forgelib", "segno>=1.6", # QR codes ] ``` Forge reads `forge.toml`. The connection string is the only required setting; the `signing_secret` turns on presigned blob URLs, which you’ll want if you later hand out direct links to stored files. forge.toml ```toml [postgres] url = "${DATABASE_URL:-postgres://postgres:forge@localhost:5432/linksapp}" [blob] signing_secret = "${BLOB_SIGNING_SECRET:-dev-secret-change-me}" ``` The app is a handful of modules: * app * main.py init Forge, start the workers, wire up FastAPI * routes.py the handlers * utils.py validation and key helpers * worker.py the background workers * pyproject.toml * forge.toml ## The app and its background loops [Section titled “The app and its background loops”](#the-app-and-its-background-loops) The whole runtime starts in a FastAPI lifespan: `init()` builds the Forge client, and three background tasks run for the life of the app, a worker per queue plus a loop that ticks the scheduler. The client is stashed on `app.state` so handlers can reach it. app/main.py ```python import asyncio from contextlib import asynccontextmanager import forgelib from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from .routes import api from .worker import clicks_worker, expire_worker, scheduler_loop @asynccontextmanager async def lifespan(app: FastAPI): forge = await forgelib.ForgeClient.init() # reads ./forge.toml, runs migrations stop = asyncio.Event() tasks = [ asyncio.create_task(clicks_worker(forge, stop)), asyncio.create_task(expire_worker(forge, stop)), asyncio.create_task(scheduler_loop(forge, stop)), ] app.state.forge = forge try: yield finally: stop.set() for t in tasks: t.cancel() await asyncio.gather(*tasks, return_exceptions=True) app = FastAPI(lifespan=lifespan) app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) app.include_router(api) @app.exception_handler(HTTPException) async def on_http_error(_req: Request, exc: HTTPException) -> JSONResponse: return JSONResponse({"error": str(exc.detail)}, status_code=exc.status_code) ``` ## Helpers [Section titled “Helpers”](#helpers) Validation, the bearer token, the key builders, and a slug generator. app/utils.py ```python import random import re import string from fastapi import HTTPException SLUG_RE = re.compile(r"^[A-Za-z0-9_-]{3,32}$") RESERVED = {"api", "healthz"} _CHARS = string.ascii_letters + string.digits def random_slug(n: int = 7) -> str: return "".join(random.choices(_CHARS, k=n)) def bearer_token(authorization: str | None) -> str: if not authorization: raise HTTPException(401, "authentication required") token = authorization.removeprefix("Bearer ").strip() if not token: raise HTTPException(401, "authentication required") return token def validate_url(raw: str) -> str: url = raw.strip() if not url.startswith(("http://", "https://")) or len(url) > 2048: raise HTTPException(400, "enter a valid http(s) url") return url def user_email_key(email: str) -> str: return f"link:user:email:{email}" def user_id_key(uid: str) -> str: return f"link:user:id:{uid}" def slug_key(slug: str) -> str: return f"link:slug:{slug}" def owner_key(uid: str) -> str: return f"link:owner:{uid}" def clicks_key(slug: str) -> str: return f"clicks:{slug}" def qr_key(slug: str) -> str: return f"qr:{slug}" def click_topic(slug: str) -> str: return f"clicks:{slug}" ``` ## Accounts [Section titled “Accounts”](#accounts) Sign-up throttles by email, hashes the password, and claims the email with a “set if not present” write that returns `False` when it’s taken. Login verifies the password and starts a session. Protected routes validate the bearer token up front. app/routes.py ```python import io import json import uuid from datetime import UTC, datetime, timedelta from typing import Annotated, Any import segno from fastapi import APIRouter, Header, HTTPException, Request, Response from fastapi.responses import RedirectResponse, StreamingResponse from pydantic import BaseModel from .utils import (RESERVED, SLUG_RE, bearer_token, click_topic, clicks_key, owner_key, qr_key, random_slug, slug_key, user_email_key, user_id_key, validate_url) from .worker import delete_link api = APIRouter() AuthH = Annotated[str | None, Header(alias="Authorization")] class Credentials(BaseModel): email: str password: str class LinkCreate(BaseModel): url: str slug: str | None = None ttl_seconds: int | None = None @api.post("/api/signup", status_code=201) async def signup(request: Request, body: Credentials) -> dict[str, Any]: forge = request.app.state.forge email = body.email.strip().lower() limit = await forge.rate_limit_check("links-auth", email, 20, 60.0, True) if not limit.allowed: raise HTTPException(429, "too many attempts; slow down") user = {"id": str(uuid.uuid4()), "email": email, "password_hash": await forge.hash_password(body.password)} body_json = json.dumps(user) if not await forge.kv_set(user_email_key(email), body_json, None, True): raise HTTPException(409, "email already registered") await forge.kv_set(user_id_key(user["id"]), body_json) token = await forge.create_session(user["id"], 1800.0, 604800.0) return {"token": token, "user": {"id": user["id"], "email": email}} @api.post("/api/login") async def login(request: Request, body: Credentials) -> dict[str, Any]: forge = request.app.state.forge email = body.email.strip().lower() limit = await forge.rate_limit_check("links-auth", email, 20, 60.0, True) if not limit.allowed: raise HTTPException(429, "too many attempts; slow down") raw = await forge.kv_get(user_email_key(email)) if raw is None: raise HTTPException(401, "invalid email or password") user = json.loads(raw) if not await forge.verify_password(body.password, user["password_hash"]): raise HTTPException(401, "invalid email or password") token = await forge.create_session(user["id"], 1800.0, 604800.0) return {"token": token, "user": {"id": user["id"], "email": email}} async def require_user(forge, authorization: str | None) -> str: user_id = await forge.validate_session(bearer_token(authorization)) if user_id is None: raise HTTPException(401, "authentication required") return user_id ``` ## Create a link [Section titled “Create a link”](#create-a-link) This is where five primitives come together. A feature flag decides whether this user may pick a custom slug. A config value supplies the per-user quota. Reserving the slug is a “set if not present” write, so the slug key is its own collision check, with a retry loop for the random case. Then store a QR image in blob, and if the link has a TTL, schedule its deletion. app/routes.py (continued) ```python DEFAULT_MAX_LINKS = 100 @api.post("/api/links", status_code=201) async def create_link(request: Request, body: LinkCreate, authorization: AuthH = None) -> dict[str, Any]: forge = request.app.state.forge user_id = await require_user(forge, authorization) url = validate_url(body.url) # config: per-user quota, with a code default when unset raw_max = await forge.config_get("max_links_per_user") max_links = int(raw_max) if raw_max else DEFAULT_MAX_LINKS owned = json.loads(await forge.kv_get(owner_key(user_id)) or "[]") if len(owned) >= max_links: raise HTTPException(409, "link limit reached") # feature flag: custom slugs, evaluated for this user custom_on = await forge.flag("custom_slugs", False, user_id) now = datetime.now(UTC).isoformat() expires_at = None if body.ttl_seconds and body.ttl_seconds > 0: expires_at = (datetime.now(UTC) + timedelta(seconds=body.ttl_seconds)).isoformat() def record(s: str) -> str: return json.dumps({"slug": s, "url": url, "ownerId": user_id, "createdAt": now, "expiresAt": expires_at}) # reserve the slug; a "set if not present" miss means it's taken if body.slug and custom_on: slug = body.slug if not SLUG_RE.match(slug) or slug in RESERVED: raise HTTPException(400, "invalid slug") if not await forge.kv_set(slug_key(slug), record(slug), None, True): raise HTTPException(409, "slug already taken") else: for _ in range(5): slug = random_slug() if await forge.kv_set(slug_key(slug), record(slug), None, True): break else: raise HTTPException(409, "could not allocate a slug") owned.insert(0, {"slug": slug, "url": url, "createdAt": now, "expiresAt": expires_at}) await forge.kv_set(owner_key(user_id), json.dumps(owned)) # render a QR image and store it in blob buf = io.BytesIO() segno.make(f"/{slug}", error="m").save(buf, kind="svg", scale=4, border=1) await forge.blob_put(qr_key(slug), buf.getvalue(), "image/svg+xml") # schedule a one-shot delete at the TTL if expires_at: when_ms = datetime.fromisoformat(expires_at).timestamp() * 1000 await forge.schedule_at(when_ms, "link-expire", json.dumps({"slug": slug})) return {"slug": slug, "url": url, "createdAt": now, "expiresAt": expires_at, "clicks": 0} ``` ## Redirect fast [Section titled “Redirect fast”](#redirect-fast) The redirect is the hot path, so keep it lean: look up the link, honor its expiry, check a per-slug rate limit so one link can’t be used to flood you, bump the atomic counter, queue the rest, and redirect. The counter increment is cheap and stays inline; everything derived from the click is deferred to the worker. app/routes.py (continued) ```python # registered last so it never shadows /api/... routes @api.get("/{slug}") async def redirect(request: Request, slug: str) -> Response: if not SLUG_RE.match(slug) or slug in RESERVED: raise HTTPException(404, "link not found") forge = request.app.state.forge raw = await forge.kv_get(slug_key(slug)) if raw is None: raise HTTPException(404, "link not found") link = json.loads(raw) if link["expiresAt"] and datetime.fromisoformat(link["expiresAt"]) <= datetime.now(UTC): raise HTTPException(404, "link not found") rl = await forge.rate_limit_check("redirect", slug, 600, 60.0, True) if not rl.allowed: raise HTTPException(429, "too many requests") await forge.kv_incr(clicks_key(slug), 1) await forge.queue("clicks").enqueue({"slug": slug}, max_attempts=3) return RedirectResponse(link["url"], status_code=302) ``` ## Live counts and the QR image [Section titled “Live counts and the QR image”](#live-counts-and-the-qr-image) The dashboard opens a server-sent events stream; the route subscribes to the slug’s topic and forwards each message, closing the subscription when the client leaves. The QR image is just a blob read. app/routes.py (continued) ```python @api.get("/api/links/{slug}/live") async def live(request: Request, slug: str) -> StreamingResponse: forge = request.app.state.forge async def events(): async for event in forge.topic(click_topic(slug)).subscribe(): yield f"data: {json.dumps(event)}\n\n" return StreamingResponse(events(), media_type="text/event-stream") @api.get("/api/links/{slug}/qr.svg") async def qr(request: Request, slug: str) -> Response: svg = await request.app.state.forge.blob_get(qr_key(slug)) if svg is None: raise HTTPException(404, "not found") return Response(content=svg, media_type="image/svg+xml") ``` ## The workers [Section titled “The workers”](#the-workers) Two queues need draining. The `clicks` worker reads the current count and publishes it so live dashboards update. The `link-expire` worker deletes a link when its scheduled TTL fires. The scheduler loop turns due `schedule_at` jobs into queue work and runs the maintenance sweep. The delete is idempotent, which is what makes at-least-once redelivery safe. app/worker.py ```python import asyncio import json from .utils import click_topic, clicks_key, owner_key, qr_key, slug_key async def delete_link(forge, slug: str) -> None: raw = await forge.kv_get(slug_key(slug)) if raw is None: return # already gone; idempotent owner_id = json.loads(raw).get("ownerId") if owner_id: owned = json.loads(await forge.kv_get(owner_key(owner_id)) or "[]") owned = [x for x in owned if x.get("slug") != slug] await forge.kv_set(owner_key(owner_id), json.dumps(owned)) await forge.kv_delete(slug_key(slug)) await forge.kv_delete(clicks_key(slug)) await forge.blob_delete(qr_key(slug)) async def clicks_worker(forge, stop: asyncio.Event) -> None: async def handle(job): slug = job.payload["slug"] total = int(await forge.kv_get(clicks_key(slug)) or "0") await forge.topic(click_topic(slug)).publish({"slug": slug, "clicks": total}) await forge.worker("clicks", handle, wait_seconds=1.0, stop=stop) async def expire_worker(forge, stop: asyncio.Event) -> None: async def handle(job): await delete_link(forge, job.payload["slug"]) await forge.worker("link-expire", handle, wait_seconds=5.0, stop=stop) async def scheduler_loop(forge, stop: asyncio.Event) -> None: while not stop.is_set(): try: await forge.run_scheduler_once() await forge.maintain() except Exception as exc: print("scheduler:", exc, flush=True) try: await asyncio.wait_for(stop.wait(), timeout=30.0) except TimeoutError: pass ``` ## Run it [Section titled “Run it”](#run-it) ```sh createdb linksapp uvicorn app.main:app --reload ``` Terminal ```sh TOKEN=$(curl -s localhost:8000/api/signup -H 'content-type: application/json' \ -d '{"email":"a@b.com","password":"supersecret"}' | jq -r .token) # create a link curl -s localhost:8000/api/links -H "authorization: Bearer $TOKEN" \ -H 'content-type: application/json' -d '{"url":"https://example.com"}' # follow the slug it returns and watch the counter curl -is localhost:8000/ ``` That’s all eight primitives in one app. The chat tutorial uses the same pieces for a realtime app, in Node. [Build a chat app (Node)](/tutorials/chat-app/)Realtime messages, presence, typing, and attachments in Hono.