feat(storage): add PostgreSQL runtime support (#377)

This commit is contained in:
l1shen
2026-08-20 14:00:48 +08:00
committed by GitHub
parent 6871f2049e
commit 15166304e7
29 changed files with 2187 additions and 197 deletions
+16
View File
@@ -30,6 +30,22 @@ jobs:
ci:
name: Lint, Typecheck & Test
runs-on: ubuntu-latest
services:
postgres:
image: postgres:15-alpine
env:
POSTGRES_DB: open_connector_test
POSTGRES_PASSWORD: postgres
POSTGRES_USER: postgres
ports:
- 5432:5432
options: >-
--health-cmd "pg_isready -U postgres"
--health-interval 5s
--health-timeout 5s
--health-retries 10
env:
TEST_POSTGRES_URL: postgresql://postgres:[email protected]:5432/open_connector_test
steps:
- name: Checkout
uses: actions/checkout@v4
+16 -6
View File
@@ -56,8 +56,8 @@ HTTP/OpenAPI from custom clients, and the Web Console for administration and deb
source.
- Runtime controls for connection identity, scopes, runtime tokens, action allow/block policies,
temporary file transit, and redacted run logs.
- Deployment options for local Docker or Node.js with local or S3-compatible transit storage,
Fly.io with persistent SQLite storage, Cloudflare Workers with D1/R2/Static Assets, and OOMOL's
- Deployment options for local Docker or Node.js with SQLite or PostgreSQL state and local or
S3-compatible transit storage, Fly.io, Cloudflare Workers with D1/R2/Static Assets, and OOMOL's
hosted runtime.
## Where It Fits
@@ -129,8 +129,8 @@ safe account labels, and execution results needed for the run.
| Path | Best for | Includes |
| ------------------------------- | ------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Open-source self-host | Developers and teams that want full control | Local Docker or Node runtime, SQLite state, local or S3-compatible transit files, MCP, HTTP, OpenAPI, and Web Console |
| Fly.io self-host | Teams that want a hosted Docker runtime | Node Docker runtime, SQLite storage on a Fly volume, TLS, health checks, MCP, HTTP, OpenAPI, and Web Console |
| Open-source self-host | Developers and teams that want full control | Local Docker or Node runtime, SQLite or PostgreSQL state, local or S3-compatible transit files, MCP, HTTP, OpenAPI, and Web Console |
| Fly.io self-host | Teams that want a hosted Docker runtime | Node Docker runtime, SQLite on a Fly volume or external PostgreSQL, TLS, health checks, MCP, HTTP, OpenAPI, and Web Console |
| Cloudflare-compatible deploy | Teams that want a lightweight hosted runtime | Workers runtime, D1 state, R2 transit files, and Static Assets for the console |
| [OOMOL](https://oomol.com/apps) | Teams that want users to authorize accounts immediately | OOMOL-provided OAuth apps, monthly included Connect credits, and hosted runtime infrastructure; the same provider and Action contracts keep a path open to later private or self-hosted deployment |
@@ -217,6 +217,16 @@ files, and Static Assets for the Web Console.
See [docs/cloudflare.md](docs/cloudflare.md) for resource creation, migrations, secrets, local Worker
preview, and remote deployment.
## PostgreSQL Runtime Storage
The Node runtime uses SQLite by default and can use PostgreSQL 15 or newer when
`OOMOL_CONNECT_DATABASE_URL` is configured. PostgreSQL migrations are explicit: run
`npm run runtime:migrate` before starting a version with pending migrations. Server startup only
checks schema readiness and never applies PostgreSQL DDL. See
[docs/configuration.md](docs/configuration.md#runtime-database) for configuration, permissions, TLS,
and multi-instance requirements. The Docker image exposes the same runner as its `migrate`
subcommand; see [docs/docker-ghcr.md](docs/docker-ghcr.md#postgresql-migrations).
## Fly.io Deployment
OpenConnector can also run on Fly.io with the Node Docker runtime and persistent SQLite storage on a
@@ -228,8 +238,8 @@ domains, and scaling.
## Docker Image (GHCR)
Run OpenConnector from a prebuilt image on GitHub Packages (GHCR): `ghcr.io/oomol-lab/open-connector`. Use
`latest` for the newest release, a pinned version like `v1.0.0` for production, or `tip` for the
latest `main` build.
`latest` for the newest release, a pinned released version for production, or `tip` for the latest
`main` build.
See [docs/docker-ghcr.md](docs/docker-ghcr.md) for tags, pulling, and running.
+8 -6
View File
@@ -96,7 +96,7 @@ We follow a coordinated disclosure process:
- Vulnerabilities in third-party providers or their APIs. Report those to the provider.
- Insecure **self-hosted configuration** that this project documents how to avoid — for example
running without `OOMOL_CONNECT_ENCRYPTION_KEY`, running without `OOMOL_CONNECT_ADMIN_TOKEN`,
binding to `0.0.0.0` on an untrusted network, or exposing the SQLite/D1/R2 data store. See
binding to `0.0.0.0` on an untrusted network, or exposing the SQLite/PostgreSQL/D1/R2 data store. See
[Hardening your deployment](#hardening-your-deployment).
- The hosted [OOMOL](https://oomol.com/) service and other OOMOL products. These are maintained
separately from this repository and are not covered by this policy; report issues in them to
@@ -132,11 +132,13 @@ responsibility for securing their deployment. At minimum:
- **Control network exposure.** The Node server binds `127.0.0.1` by default; the Docker image binds
`0.0.0.0`. Only expose the gateway on a trusted network or behind an authenticated proxy, and never
expose it publicly without admin and runtime authentication enabled.
- **Protect the data store.** The SQLite database (local), D1 (Cloudflare), and R2 transit files
contain sensitive material even when encrypted. Restrict file permissions and access, and store
Cloudflare secrets with `wrangler secret put`. Idempotent Action responses are eligible for replay
for 24 hours, but expired records are removed opportunistically rather than by a physical-deletion
deadline.
- **Protect the data store.** The SQLite or PostgreSQL database (Node), D1 (Cloudflare), and transit
files contain sensitive material even when encrypted. Restrict file, database, and bucket access;
use TLS verification for remote PostgreSQL; keep database URLs in a secret manager; and store
Cloudflare secrets with `wrangler secret put`. PostgreSQL migrations can use a separate DDL role,
while the application role only needs runtime-table CRUD and `runtime_migrations` read access.
Idempotent Action responses are eligible for replay for 24 hours, but expired records are removed
opportunistically rather than by a physical-deletion deadline.
- **Reduce attack surface.** Use `OOMOL_CONNECT_ALLOWED_ACTIONS` / `OOMOL_CONNECT_BLOCKED_ACTIONS` to
limit which Actions can run. Restrict provider proxies separately with
`OOMOL_CONNECT_ALLOWED_PROXIES` / `OOMOL_CONNECT_BLOCKED_PROXIES`: `/v1/proxy/:service` can reach
+3
View File
@@ -9,6 +9,9 @@ services:
- connector-data:/app/data
environment:
OOMOL_CONNECT_DATA_DIR: /app/data
OOMOL_CONNECT_DATABASE_URL:
OOMOL_CONNECT_DATABASE_POOL_MAX:
OOMOL_CONNECT_DATABASE_CONNECT_TIMEOUT_MS:
OOMOL_CONNECT_ORIGIN:
OOMOL_CONNECT_ENCRYPTION_KEY:
OOMOL_CONNECT_ADMIN_TOKEN:
+4 -1
View File
@@ -19,14 +19,17 @@ ENV OOMOL_CONNECT_DATA_DIR=/app/data
COPY package.json package-lock.json ./
COPY scripts/healthcheck.ts ./scripts/healthcheck.ts
COPY scripts/ensure-generated.ts ./scripts/ensure-generated.ts
COPY scripts/runtime-data.ts ./scripts/runtime-data.ts
COPY migrations ./migrations
COPY --from=build /app/src ./src
COPY --from=build /app/catalog ./catalog
COPY --from=build /app/dist ./dist
COPY --chmod=755 docker/entrypoint.sh /usr/local/bin/open-connector
RUN npm ci --omit=dev --ignore-scripts
VOLUME ["/app/data"]
EXPOSE 3000
HEALTHCHECK --interval=30s --timeout=3s --start-period=10s --retries=3 \
CMD ["node", "scripts/healthcheck.ts"]
CMD ["node", "src/server/index.ts"]
ENTRYPOINT ["/usr/local/bin/open-connector"]
CMD ["serve"]
+20
View File
@@ -0,0 +1,20 @@
#!/bin/sh
set -eu
if [ "$#" -eq 0 ]; then
set -- serve
fi
case "$1" in
serve)
shift
exec node src/server/index.ts "$@"
;;
migrate)
shift
exec node scripts/runtime-data.ts migrate "$@"
;;
*)
exec "$@"
;;
esac
+2 -2
View File
@@ -228,8 +228,8 @@ déployer, définir un domaine personnalisé et ajuster le scaling.
## Image Docker (GHCR)
Exécutez OpenConnector depuis une image préconstruite sur GitHub Packages (GHCR) :
`ghcr.io/oomol-lab/open-connector`. Utilisez `latest` pour la dernière release, une version épinglée comme
`v1.0.0` en production, ou `tip` pour le dernier build de `main`.
`ghcr.io/oomol-lab/open-connector`. Utilisez `latest` pour la dernière release, une version de release
épinglée en production, ou `tip` pour le dernier build de `main`.
Consultez [docker-ghcr.md (anglais)](docker-ghcr.md) pour les tags d'image, le pull et l'exécution.
+1 -1
View File
@@ -214,7 +214,7 @@ Fly app 作成、volume、secret、deployment、custom domain、scaling につ
## Docker イメージ(GHCR)
事前ビルドされた Docker イメージで OpenConnector を実行できます(GitHub Packages / GHCR):
`ghcr.io/oomol-lab/open-connector`。最新 release は `latest`、production では `v1.0.0` のように version を固定、
`ghcr.io/oomol-lab/open-connector`。最新 release は `latest`、production では release version を固定、
最新の `main` build は `tip` を使います。
イメージの tag、pull、実行については [docker-ghcr.md(英語)](docker-ghcr.md) を参照してください。
+1 -1
View File
@@ -221,7 +221,7 @@ OpenConnector는 Node Docker runtime과 Fly volume의 영구 SQLite storage를
## Docker 이미지(GHCR)
GitHub Packages(GHCR)의 사전 빌드된 이미지 `ghcr.io/oomol-lab/open-connector`로 OpenConnector를
실행할 수 있습니다. 최신 release에는 `latest`, 재현 가능한 production 배포에는 `v1.0.0` 같은 고정
실행할 수 있습니다. 최신 release에는 `latest`, 재현 가능한 production 배포에는 고정된 release
version, 최신 `main` build에는 `tip`을 사용하세요.
Image tag, 가져오기, 실행 방법은 [docker-ghcr.md](docker-ghcr.md)를 참조하세요.
+2 -2
View File
@@ -225,8 +225,8 @@ Fly volume.
## Docker-образ (GHCR)
Запускайте OpenConnector из готового образа в GitHub Packages (GHCR): `ghcr.io/oomol-lab/open-connector`.
Используйте `latest` для новейшего release, закреплённую версию вроде `v1.0.0` для production или `tip`
для последнего build из `main`.
Используйте `latest` для новейшего release, закреплённую версию release для production или `tip` для
последнего build из `main`.
О тегах образа, pull и запуске см. [docker-ghcr.md (на английском)](docker-ghcr.md).
+1 -1
View File
@@ -206,7 +206,7 @@ Fly app 创建、volume、secret、部署、自定义域名和扩缩容步骤见
## Docker 镜像(GHCR)
可以直接用 GitHub Packages(GHCR)上的预构建镜像运行 OpenConnector:`ghcr.io/oomol-lab/open-connector`。最新
release 用 `latest`,生产环境固定版本号(如 `v1.0.0`),想用最新 `main` 构建则用 `tip`。
release 用 `latest`,生产环境固定具体的 release 版本号,想用最新 `main` 构建则用 `tip`。
镜像标签、拉取和运行的说明见 [docker-ghcr.zh-CN.md](docker-ghcr.zh-CN.md)。
+1 -1
View File
@@ -203,7 +203,7 @@ OpenConnector 也可部署至 Fly.io,使用 Node Docker 執行階段,並將
## Docker 映像檔(GHCR)
使用 GitHub Packages(GHCR)上的預先建置映像檔執行 OpenConnector:`ghcr.io/oomol-lab/open-connector`。
最新版本使用 `latest`;正式環境請固定版本號,例如 `v1.0.0`;若要使用最新的 `main` 建置則使用 `tip`。
最新版本使用 `latest`;正式環境請固定具體的 release 版本號;若要使用最新的 `main` 建置則使用 `tip`。
標籤、拉取及執行方式,請參閱 [docker-ghcr.md(英文)](docker-ghcr.md)。
+73 -33
View File
@@ -2,38 +2,41 @@
OpenConnector is configured with environment variables.
| Variable | Default | Purpose |
| ---------------------------------------- | ------------------------- | --------------------------------------------------------------------------------------------------- |
| `PORT` | `3000` | Local HTTP server port. |
| `HOST` | `127.0.0.1` | Bind address. Docker image sets `0.0.0.0`. |
| `OOMOL_CONNECT_ORIGIN` | `http://localhost:<PORT>` | Public origin used for OAuth redirect URLs. |
| `OOMOL_CONNECT_DATA_DIR` | `./data` | Runtime database, local transit files, and Node upload staging. Docker uses `/app/data`. |
| `OOMOL_CONNECT_ENCRYPTION_KEY` | unset | Encrypts credentials, OAuth config, pending OAuth state, and completed idempotent Action responses. |
| `OOMOL_CONNECT_NEW_ENCRYPTION_KEY` | unset | New key used by `runtime:data rotate-key`. |
| `OOMOL_CONNECT_ADMIN_TOKEN` | unset | Requires bearer-token auth for local admin API, docs, and web console. |
| `OOMOL_CONNECT_RUNTIME_TOKEN` | unset | Optional bootstrap runtime bearer token for `/v1` and MCP callers. |
| `OOMOL_CONNECT_ALLOWED_CUSTOM_OAUTH` | unset | Enables connection-scoped OAuth apps for `*` or a comma-separated service list. |
| `OOMOL_CONNECT_JWKS_URI` | unset | Node-only JWKS endpoint for validating runtime JWT access tokens. |
| `OOMOL_CONNECT_JWT_ISSUER` | unset | Expected `iss` claim for runtime JWT access tokens. |
| `OOMOL_CONNECT_JWT_AUDIENCE` | unset | Expected API `aud` claim for runtime JWT access tokens. |
| `OOMOL_CONNECT_ALLOWED_ACTIONS` | unset | Comma-separated executable action allowlist. Supports `service.*` and `*`. |
| `OOMOL_CONNECT_BLOCKED_ACTIONS` | unset | Comma-separated executable action denylist. Supports `service.*` and `*`. |
| `OOMOL_CONNECT_ALLOWED_PROXIES` | unset | Comma-separated provider proxy allowlist. Supports service names and `*`. |
| `OOMOL_CONNECT_BLOCKED_PROXIES` | unset | Comma-separated provider proxy denylist. Supports service names and `*`. |
| `OOMOL_CONNECT_ALLOW_PRIVATE_NETWORK` | `false` | Allow self-hosted provider connections to target private networks. See below. |
| `OOMOL_CONNECT_EGRESS_TRUSTED_HOSTS` | unset | Trusted hosts routed through a corporate VPN. See below. |
| `OOMOL_CONNECT_LOG_LEVEL` | `info` | Pino log level for the local Node server. |
| `OOMOL_CONNECT_TRANSIT_FILE_BACKEND` | `local` | Node transit-file backend: `local` or `s3`. |
| `OOMOL_CONNECT_TRANSIT_FILE_TTL_SECONDS` | `86400` | Transit file lifetime before cleanup. |
| `OOMOL_CONNECT_TRANSIT_FILE_MAX_BYTES` | `104857600` | Maximum transit file upload size. |
| `OOMOL_CONNECT_S3_BUCKET` | unset | Bucket used when the Node transit-file backend is `s3`. |
| `OOMOL_CONNECT_S3_REGION` | `us-east-1` | S3 signing region. Use the value required by the storage service. |
| `OOMOL_CONNECT_S3_ENDPOINT` | AWS S3 | S3-compatible endpoint for MinIO, Cloudflare R2, Alibaba Cloud OSS, or another object store. |
| `OOMOL_CONNECT_S3_FORCE_PATH_STYLE` | `false` | Use path-style S3 URLs. This is commonly required by local MinIO deployments. |
| `OOMOL_CONNECT_S3_ACCESS_KEY_ID` | SDK credential chain | Explicit S3 access key ID. Configure it with the secret key, or omit both. |
| `OOMOL_CONNECT_S3_SECRET_ACCESS_KEY` | SDK credential chain | Explicit S3 secret access key. Configure it with the access key ID, or omit both. |
| `OOMOL_CONNECT_S3_SESSION_TOKEN` | unset | Optional session token used with explicit S3 credentials. |
| `OOMOL_CONNECT_RUN_LIMIT` | `5000` | Maximum number of recent action run audit records to retain. |
| Variable | Default | Purpose |
| ------------------------------------------- | ------------------------- | --------------------------------------------------------------------------------------------------- |
| `PORT` | `3000` | Local HTTP server port. |
| `HOST` | `127.0.0.1` | Bind address. Docker image sets `0.0.0.0`. |
| `OOMOL_CONNECT_ORIGIN` | `http://localhost:<PORT>` | Public origin used for OAuth redirect URLs. |
| `OOMOL_CONNECT_DATA_DIR` | `./data` | SQLite database, local transit files, and Node upload staging. Docker uses `/app/data`. |
| `OOMOL_CONNECT_DATABASE_URL` | unset | PostgreSQL connection URL. When unset, the Node runtime uses SQLite under the data directory. |
| `OOMOL_CONNECT_DATABASE_POOL_MAX` | `10` | Maximum PostgreSQL connections per Node runtime instance. |
| `OOMOL_CONNECT_DATABASE_CONNECT_TIMEOUT_MS` | `10000` | PostgreSQL connection timeout in milliseconds. |
| `OOMOL_CONNECT_ENCRYPTION_KEY` | unset | Encrypts credentials, OAuth config, pending OAuth state, and completed idempotent Action responses. |
| `OOMOL_CONNECT_NEW_ENCRYPTION_KEY` | unset | New key used by `runtime:data rotate-key`. |
| `OOMOL_CONNECT_ADMIN_TOKEN` | unset | Requires bearer-token auth for local admin API, docs, and web console. |
| `OOMOL_CONNECT_RUNTIME_TOKEN` | unset | Optional bootstrap runtime bearer token for `/v1` and MCP callers. |
| `OOMOL_CONNECT_ALLOWED_CUSTOM_OAUTH` | unset | Enables connection-scoped OAuth apps for `*` or a comma-separated service list. |
| `OOMOL_CONNECT_JWKS_URI` | unset | Node-only JWKS endpoint for validating runtime JWT access tokens. |
| `OOMOL_CONNECT_JWT_ISSUER` | unset | Expected `iss` claim for runtime JWT access tokens. |
| `OOMOL_CONNECT_JWT_AUDIENCE` | unset | Expected API `aud` claim for runtime JWT access tokens. |
| `OOMOL_CONNECT_ALLOWED_ACTIONS` | unset | Comma-separated executable action allowlist. Supports `service.*` and `*`. |
| `OOMOL_CONNECT_BLOCKED_ACTIONS` | unset | Comma-separated executable action denylist. Supports `service.*` and `*`. |
| `OOMOL_CONNECT_ALLOWED_PROXIES` | unset | Comma-separated provider proxy allowlist. Supports service names and `*`. |
| `OOMOL_CONNECT_BLOCKED_PROXIES` | unset | Comma-separated provider proxy denylist. Supports service names and `*`. |
| `OOMOL_CONNECT_ALLOW_PRIVATE_NETWORK` | `false` | Allow self-hosted provider connections to target private networks. See below. |
| `OOMOL_CONNECT_EGRESS_TRUSTED_HOSTS` | unset | Trusted hosts routed through a corporate VPN. See below. |
| `OOMOL_CONNECT_LOG_LEVEL` | `info` | Pino log level for the local Node server. |
| `OOMOL_CONNECT_TRANSIT_FILE_BACKEND` | `local` | Node transit-file backend: `local` or `s3`. |
| `OOMOL_CONNECT_TRANSIT_FILE_TTL_SECONDS` | `86400` | Transit file lifetime before cleanup. |
| `OOMOL_CONNECT_TRANSIT_FILE_MAX_BYTES` | `104857600` | Maximum transit file upload size. |
| `OOMOL_CONNECT_S3_BUCKET` | unset | Bucket used when the Node transit-file backend is `s3`. |
| `OOMOL_CONNECT_S3_REGION` | `us-east-1` | S3 signing region. Use the value required by the storage service. |
| `OOMOL_CONNECT_S3_ENDPOINT` | AWS S3 | S3-compatible endpoint for MinIO, Cloudflare R2, Alibaba Cloud OSS, or another object store. |
| `OOMOL_CONNECT_S3_FORCE_PATH_STYLE` | `false` | Use path-style S3 URLs. This is commonly required by local MinIO deployments. |
| `OOMOL_CONNECT_S3_ACCESS_KEY_ID` | SDK credential chain | Explicit S3 access key ID. Configure it with the secret key, or omit both. |
| `OOMOL_CONNECT_S3_SECRET_ACCESS_KEY` | SDK credential chain | Explicit S3 secret access key. Configure it with the access key ID, or omit both. |
| `OOMOL_CONNECT_S3_SESSION_TOKEN` | unset | Optional session token used with explicit S3 credentials. |
| `OOMOL_CONNECT_RUN_LIMIT` | `5000` | Maximum number of recent action run audit records to retain. |
Example:
@@ -47,7 +50,8 @@ npm run dev
```
Create persistent runtime tokens from the web console Access tab or `POST /api/runtime-tokens`.
Only token hashes are stored: the Node server stores persistent-token records in SQLite, while
Only token hashes are stored: the Node server stores persistent-token records in its selected SQLite
or PostgreSQL database, while
Cloudflare Workers store them in D1. Persistent tokens have independent Action rules, provider
proxy grants, and optional connection grants. A new token has no proxy access until its
`allowedProxies` includes a provider service or `*`; those grants can only narrow the deployment
@@ -61,6 +65,42 @@ connections do not require a grant. Create requests may omit the field; updates
Because the bootstrap token has no stored policy, its proxy access is controlled by the deployment
and runtime rules, while its connection access remains unrestricted.
## Runtime database
The Node runtime uses `OOMOL_CONNECT_DATA_DIR/connect.sqlite` by default and applies SQLite
migrations automatically when it opens the database. Leave `OOMOL_CONNECT_DATABASE_URL` unset to
keep this zero-configuration mode.
Set a `postgres:` or `postgresql:` connection URL to use PostgreSQL 15 or newer instead:
```bash
OOMOL_CONNECT_DATABASE_URL="postgresql://open_connector:password@db.example.com:5432/open_connector?sslmode=verify-full" \
npm run runtime:migrate
OOMOL_CONNECT_DATABASE_URL="postgresql://open_connector:password@db.example.com:5432/open_connector?sslmode=verify-full" \
npm start
```
PostgreSQL migrations are explicit: run `npm run runtime:migrate` before the first server start and
before starting a new application version that includes pending migrations. The server performs a
read-only schema check and refuses to start with an actionable error when required migrations are
missing. It never applies PostgreSQL DDL during startup.
The migration command and application may use different database roles. The migration role needs
permission to create and alter runtime tables and indexes. The application role needs CRUD access to
the runtime tables plus `SELECT` access to `runtime_migrations`; it does not need DDL permission.
Treat both connection URLs as secrets and use a TLS-verifying connection configuration for remote
databases.
Each process selects exactly one backend. PostgreSQL and SQLite are not synchronized, and selecting
PostgreSQL does not import an existing `connect.sqlite` database. Removing the URL makes the runtime
use the existing SQLite database again.
For multiple Node instances, configure the same PostgreSQL database and
`OOMOL_CONNECT_ENCRYPTION_KEY` on every instance. Also use the S3-compatible transit-file backend so
files are shared across instances. Size `OOMOL_CONNECT_DATABASE_POOL_MAX` so the total across all
instances leaves capacity for migrations and database administration.
## JWT access tokens
The Node server can validate JWT access tokens issued by an existing identity provider for `/v1/*`
+26 -20
View File
@@ -1,8 +1,9 @@
# Credentials And Local Storage
# Credentials And Runtime Storage
The local Node runtime stores connections, OAuth client configuration, pending OAuth states, runtime
tokens, recent run logs, and HTTP Action idempotency claims and responses in SQLite. The Cloudflare
Workers runtime stores the same runtime records in D1 and temporary transit files in R2.
The Node runtime stores connections, OAuth client configuration, pending OAuth states, runtime
tokens, recent run logs, and HTTP Action idempotency claims and responses in SQLite or PostgreSQL.
The Cloudflare Workers runtime stores the same runtime records in D1 and temporary transit files in
R2.
By default the database lives at:
@@ -11,10 +12,11 @@ By default the database lives at:
```
Set `OOMOL_CONNECT_DATA_DIR` to use another directory. The Docker image defaults this to
`/app/data`, which is intended to be mounted as a volume.
`/app/data`, which is intended to be mounted as a volume. Set `OOMOL_CONNECT_DATABASE_URL` to use a
PostgreSQL database instead; see [configuration](configuration.md#runtime-database).
- `no_auth` providers are available as virtual connections and do not store secrets.
- `api_key` and `custom_credential` providers store their local secrets in SQLite.
- `api_key` and `custom_credential` providers store their local secrets in the selected runtime database.
- `oauth2` providers use user-provided OAuth client configuration and a runtime callback URL.
## Encryption
@@ -35,8 +37,8 @@ stored by OpenConnector; if it is lost, encrypted records cannot be recovered.
Without `OOMOL_CONNECT_ENCRYPTION_KEY`, the runtime stays usable for local development and prints a
startup warning. In that mode, credentials, OAuth client configuration, pending OAuth state, and
completed idempotent Action responses are stored as plaintext. Action responses may contain sensitive provider data, so
treat `connect.sqlite` or D1 as a sensitive data store even after a response is no longer eligible
for replay.
treat `connect.sqlite`, PostgreSQL, or D1 as a sensitive data store even after a response is no
longer eligible for replay.
Completed idempotent Action responses remain eligible for replay for 24 hours. Expired idempotency
records are deleted opportunistically when a later idempotent Action request claims a key; the
@@ -208,7 +210,7 @@ curl -s -X POST http://localhost:3000/api/oauth/authorizations \
-d '{"service":"github","connectionName":"work"}'
```
Protect the local SQLite database like any other file containing API keys or OAuth tokens.
Protect the selected runtime database like any other store containing API keys or OAuth tokens.
## Selecting A Connection For Execution
@@ -267,13 +269,13 @@ curl -s -X POST http://localhost:3000/api/runtime-tokens \
## Reset And Key Rotation
Reset local runtime data:
Reset Node runtime data in the selected SQLite or PostgreSQL database:
```bash
npm run runtime:data -- reset --yes
```
Rotate the local SQLite data-encryption key:
Rotate the Node runtime data-encryption key:
```bash
OOMOL_CONNECT_ENCRYPTION_KEY="old-secret" \
@@ -281,25 +283,28 @@ OOMOL_CONNECT_NEW_ENCRYPTION_KEY="new-secret" \
npm run runtime:data -- rotate-key
```
Remove local SQLite data encryption only when you intentionally want plaintext local storage:
Remove Node runtime data encryption only when you intentionally want plaintext storage:
```bash
OOMOL_CONNECT_ENCRYPTION_KEY="old-secret" \
npm run runtime:data -- rotate-key --plain
```
Both commands re-encode stored credentials, OAuth client configuration, and completed idempotent
Action response payloads. Idempotency key hashes, request fingerprints, claim state, and timestamps
remain unencrypted metadata.
For PostgreSQL, set `OOMOL_CONNECT_DATABASE_URL` on the reset or rotate command. The schema must
already be initialized with `npm run runtime:migrate`. Key rotation re-encodes stored credentials,
OAuth client configuration, pending OAuth state, and completed idempotent Action response payloads.
Idempotency key hashes, request fingerprints, claim state, and timestamps remain unencrypted
metadata. Pause all Node runtime instances before resetting data or rotating a PostgreSQL encryption
key.
`runtime:data` is for the local SQLite runtime only. For Cloudflare, back up and restore D1/R2
directly with Cloudflare tooling.
`runtime:data` supports the Node SQLite and PostgreSQL backends. For Cloudflare, back up and restore
D1/R2 directly with Cloudflare tooling.
## OAuth Token Refresh
OAuth access tokens are refreshed automatically when they are expired and the provider issued a
refresh token. Refreshed credentials are written back to the local SQLite store, using encryption
when `OOMOL_CONNECT_ENCRYPTION_KEY` is configured.
refresh token. Refreshed credentials are written back to the selected Node runtime database, using
encryption when `OOMOL_CONNECT_ENCRYPTION_KEY` is configured.
If a token is expired and no refresh token is available, reconnect the provider from the local
runtime. Providers such as Google may require authorization parameters that request offline access;
@@ -323,7 +328,8 @@ Authorization: Bearer replace-with-an-admin-token
```
Create runtime tokens for `/v1` and `/mcp` callers from the web console Access tab or
`POST /api/runtime-tokens`. The token is shown once when created; only a hash is stored in SQLite.
`POST /api/runtime-tokens`. The token is shown once when created; only a hash is stored in the
selected runtime database.
Runtime clients should send `Authorization: Bearer oct_...`. Persistent tokens configure Action
rules, provider proxy grants, and optional connection grants independently. Their `allowedProxies`
list is empty by default, which denies `/v1/proxy/:service`; add a provider service or `*` only when
+35 -7
View File
@@ -11,14 +11,14 @@ ghcr.io/oomol-lab/open-connector
## Choose A Tag
| Tag | Points at | Use it when |
| ------------- | ------------------------------------ | -------------------------------------------------------------- |
| `latest` | the newest published release | you want the current stable runtime |
| `v1.0.0` | a specific release (immutable) | you deploy to production and want a pinned, reproducible build |
| `tip` | the latest commit on `main` | you want to try changes that are not released yet |
| `<short-sha>` | a specific `main` commit (immutable) | you want to pin an exact pre-release build |
| Tag | Points at | Use it when |
| ------------------- | ------------------------------------ | -------------------------------------------------------------- |
| `latest` | the newest published release | you want the current stable runtime |
| `<release-version>` | a specific release (immutable) | you deploy to production and want a pinned, reproducible build |
| `tip` | the latest commit on `main` | you want to try changes that are not released yet |
| `<short-sha>` | a specific `main` commit (immutable) | you want to pin an exact pre-release build |
For production, pin a released version such as `v1.0.0`.
For production, pin a specific released version instead of `latest`.
## Pull
@@ -70,6 +70,28 @@ docker run -d \
See [configuration.md](configuration.md) for the full environment variable reference and
[credentials.md](credentials.md) for connecting providers.
### PostgreSQL Migrations
When using PostgreSQL, run the image's explicit `migrate` command before the first server start and
before starting a newer image that contains pending migrations. Use the same image tag that you
will deploy:
```bash
OPEN_CONNECTOR_VERSION="<release-version>"
docker run --rm \
-e OOMOL_CONNECT_DATABASE_URL="postgresql://migration_user:password@db.example.com:5432/open_connector?sslmode=verify-full" \
"ghcr.io/oomol-lab/open-connector:${OPEN_CONNECTOR_VERSION}" \
migrate
```
Replace `<release-version>` with the exact release tag used by the server you will deploy. The
migration image and server image must use the same tag.
The command exits after applying migrations and does not start the HTTP server. Starting the image
without a command still starts the server, which only checks schema readiness and never applies
PostgreSQL DDL.
### Docker Compose
The repository ships a [`docker-compose.yml`](../docker-compose.yml) that runs this published image.
@@ -79,6 +101,12 @@ From a checkout, export the secrets shown above and start it:
docker compose up
```
With `OOMOL_CONNECT_DATABASE_URL` exported, run PostgreSQL migrations as a one-off Compose command:
```bash
docker compose run --rm connector migrate
```
To build from source instead of pulling, add the build overlay:
```bash
+32 -7
View File
@@ -11,14 +11,14 @@ ghcr.io/oomol-lab/open-connector
## 选择标签(Tag)
| 标签 | 指向 | 适用场景 |
| ------------- | -------------------------------- | -------------------------------------- |
| `latest` | 最新发布的 release | 想要当前的稳定 runtime |
| `v1.0.0` | 某个具体 release(不可变) | 部署到生产环境,需要固定、可复现的构建 |
| `tip` | `main` 上的最新 commit | 想体验尚未发布的改动 |
| `<short-sha>` | 某个具体 `main` commit(不可变) | 想固定到某个确切的预发布构建 |
| 标签 | 指向 | 适用场景 |
| ------------------- | -------------------------------- | -------------------------------------- |
| `latest` | 最新发布的 release | 想要当前的稳定 runtime |
| `<release-version>` | 某个具体 release(不可变) | 部署到生产环境,需要固定、可复现的构建 |
| `tip` | `main` 上的最新 commit | 想体验尚未发布的改动 |
| `<short-sha>` | 某个具体 `main` commit(不可变) | 想固定到某个确切的预发布构建 |
生产环境请固定到某个 release 版本,例如 `v1.0.0`。
生产环境请固定到某个具体的 release 版本,不要使用 `latest`。
## 拉取
@@ -66,6 +66,25 @@ docker run -d \
完整环境变量参考见 [configuration.md](configuration.md),连接 provider 见 [credentials.md](credentials.md)。
### PostgreSQL Migration
使用 PostgreSQL 时,请在服务器首次启动前,以及启动包含待执行 migration 的新镜像版本前,显式运行镜像的
`migrate` 命令。Migration 应使用即将部署的同一个镜像标签:
```bash
OPEN_CONNECTOR_VERSION="<release-version>"
docker run --rm \
-e OOMOL_CONNECT_DATABASE_URL="postgresql://migration_user:password@db.example.com:5432/open_connector?sslmode=verify-full" \
"ghcr.io/oomol-lab/open-connector:${OPEN_CONNECTOR_VERSION}" \
migrate
```
将 `<release-version>` 替换为实际部署的 release tag。Migration 镜像和服务器镜像必须使用同一个 tag。
该命令应用 migration 后便会退出,不会启动 HTTP 服务器。不给镜像传命令时仍然默认启动服务器;服务器只检查
schema 是否就绪,不会执行 PostgreSQL DDL。
### Docker Compose
仓库自带一个 [`docker-compose.yml`](../docker-compose.yml),直接运行这个发布镜像。在仓库目录下,先 export
@@ -75,6 +94,12 @@ docker run -d \
docker compose up
```
导出 `OOMOL_CONNECT_DATABASE_URL` 后,可以通过一次性 Compose 命令执行 PostgreSQL migration:
```bash
docker compose run --rm connector migrate
```
想改为从源码构建而不是拉取:
```bash
+39 -7
View File
@@ -1,8 +1,9 @@
# Fly.io Deployment
OpenConnector can run on Fly.io as the Node Docker runtime with persistent SQLite storage. Fly
provides TLS termination, remote Docker builds, health checks, rolling deploys, and optional custom
domains.
OpenConnector can run on Fly.io as the Node Docker runtime. The repository defaults to persistent
SQLite storage; external PostgreSQL and S3-compatible transit storage are available for
multi-machine deployments. Fly provides TLS termination, remote Docker builds, health checks,
rolling deploys, and optional custom domains.
This deployment uses the repository's `docker/Dockerfile`, the Fly app config in `fly.toml`, and a
Fly volume mounted at `/app/data`.
@@ -64,7 +65,7 @@ fly secrets set \
Keep `OOMOL_CONNECT_ENCRYPTION_KEY` in a password manager or another external secrets vault. If the
key is lost, encrypted credentials, OAuth client configuration, and completed idempotent Action
responses in the SQLite database cannot be recovered.
responses in the runtime database cannot be recovered.
Optional runtime policy can also be set as secrets:
@@ -92,6 +93,34 @@ The Fly config uses:
- `/health` for HTTP health checks.
- `/app/data` as the mounted persistent data directory.
## PostgreSQL And Multiple Machines
To use external PostgreSQL, set `OOMOL_CONNECT_DATABASE_URL` as a Fly secret. PostgreSQL migrations
are never applied by application startup, so run them from a trusted deployment environment before
the first deploy and before deploying a version with new migrations:
```bash
OOMOL_CONNECT_DATABASE_URL="postgresql://migrator:password@db.example.com/open_connector?sslmode=verify-full" \
npm run runtime:migrate
fly secrets set \
OOMOL_CONNECT_DATABASE_URL="postgresql://app:password@db.example.com/open_connector?sslmode=verify-full" \
--app my-open-connector
```
The migration and application URLs may use separate PostgreSQL roles. Application instances only
check migration readiness and refuse to start when a required migration is missing.
Multiple machines also need shared transit-file storage. Configure the S3 backend and the same
bucket, endpoint, credentials, and `OOMOL_CONNECT_ENCRYPTION_KEY` on every machine. See
[configuration.md](configuration.md#s3-compatible-transit-files) for the required settings.
The default `fly.toml` mount exists for SQLite. For a PostgreSQL and S3 deployment, remove the
`[mounts]` section from the deployment config so Fly does not require a per-machine volume. The
container filesystem still provides temporary upload staging, but no runtime state depends on that
filesystem. Size `OOMOL_CONNECT_DATABASE_POOL_MAX` so the combined pools of all machines leave
database capacity for migrations and administration.
## Verify The Runtime
Check the health endpoint:
@@ -155,7 +184,7 @@ fly certs check api.example.com --app my-open-connector
## Updating
Deploy new versions from the repository root:
For SQLite, deploy new versions from the repository root:
```bash
git pull
@@ -164,6 +193,8 @@ fly deploy --config fly.toml --remote-only
The mounted volume keeps `connect.sqlite` and transit files across deployments.
For PostgreSQL, run `npm run runtime:migrate` with the migration database URL before `fly deploy`.
## Scaling
`fly.toml` defaults to suspending the single machine when idle:
@@ -181,5 +212,6 @@ min_machines_running = 1
```
Keep the machine count at one for the default SQLite deployment. Fly volumes are attached to
individual machines, so horizontal scaling requires a separate shared storage design. For this
repository's default Fly setup, prefer increasing the VM size before adding more machines.
individual machines, so horizontal scaling requires PostgreSQL and shared S3-compatible transit
storage as described above. For the default Fly setup, prefer increasing the VM size before adding
more machines.
+16 -1
View File
@@ -160,6 +160,21 @@ OOMOL_CONNECT_DATA_DIR=/path/to/data npm run dev
With Docker Compose, the bundled `connector-data` volume is mounted at `/app/data`.
To use PostgreSQL 15 or newer, explicitly apply migrations before the first start and before each
upgrade that adds migrations:
```bash
OOMOL_CONNECT_DATABASE_URL="postgresql://open_connector:password@localhost:5432/open_connector" \
npm run runtime:migrate
OOMOL_CONNECT_DATABASE_URL="postgresql://open_connector:password@localhost:5432/open_connector" \
npm run dev
```
The Node server checks that all required PostgreSQL migrations are present but never applies DDL at
startup. SQLite remains the default when `OOMOL_CONNECT_DATABASE_URL` is unset; the two backends are
not synchronized.
Set `OOMOL_CONNECT_ENCRYPTION_KEY` to encrypt stored credentials, OAuth client configuration, and
completed idempotent Action responses:
@@ -177,7 +192,7 @@ curl -s http://localhost:3000/api/actions \
Use the admin token for `/api`, `/docs`, and the web console. Create persistent runtime tokens for
`/v1` and `/mcp` from the web console Access tab or `POST /api/runtime-tokens`; only token hashes are
stored in SQLite. Persistent tokens have no provider proxy access unless their independent
stored in the selected runtime database. Persistent tokens have no provider proxy access unless their independent
`allowedProxies` grant includes the provider service or `*`. `OOMOL_CONNECT_RUNTIME_TOKEN` remains
available for bootstrap scripts.
+72
View File
@@ -0,0 +1,72 @@
create table connections (
id text not null unique,
revision text not null,
service text not null,
connection_name text not null,
value text not null,
updated_at text not null,
primary key (service, connection_name)
);
create table oauth_client_configs (
service text primary key,
value text not null,
updated_at text not null
);
create table oauth_states (
state text primary key,
value text not null,
created_at text not null
);
create table runtime_tokens (
id text primary key,
name text not null,
token_hash text not null unique,
allowed_actions text not null default '[]',
blocked_actions text not null default '[]',
allowed_proxies text not null default '[]',
created_at text not null,
last_used_at text,
revoked_at text
);
create table runtime_policy (
id integer primary key check (id = 1),
value text not null,
updated_at text not null
);
create table runs (
id text primary key,
service text not null,
action_id text not null,
caller text not null,
started_at text not null,
completed_at text not null,
ok integer not null check (ok in (0, 1)),
value text not null
);
create index runs_started_at_id_idx on runs (started_at desc, id desc);
create index runs_service_started_at_id_idx on runs (service, started_at desc, id desc);
create index runs_action_id_started_at_id_idx on runs (action_id, started_at desc, id desc);
create index runs_caller_started_at_id_idx on runs (caller, started_at desc, id desc);
create index runs_ok_started_at_id_idx on runs (ok, started_at desc, id desc);
create table idempotency_records (
key_hash text primary key,
claim_id text not null,
request_hash text not null,
state text not null check (state in ('in_progress', 'completed')),
response_value text,
created_at text not null,
expires_at text not null,
check (
(state = 'in_progress' and response_value is null)
or (state = 'completed' and response_value is not null)
)
);
create index idempotency_records_expires_at_idx on idempotency_records (expires_at);
@@ -0,0 +1 @@
alter table runtime_tokens add column allowed_connections text not null default '[]';
+248
View File
@@ -34,11 +34,14 @@
"minisearch": "^7.2.0",
"mqtt": "^5.15.2",
"nodemailer": "^9.0.3",
"pg": "^8.23.0",
"pino": "^10.3.1",
"rss-parser": "^3.13.0",
"zod": "^4.4.3"
},
"devDependencies": {
"@electric-sql/pglite": "^0.5.5",
"@electric-sql/pglite-socket": "^0.2.8",
"@hyrious/configs": "^0.1.6",
"@types/ali-oss": "^6.23.3",
"@types/busboy": "^1.5.4",
@@ -46,6 +49,7 @@
"@types/mdast": "^4.0.4",
"@types/node": "^26.0.1",
"@types/nodemailer": "^8.0.1",
"@types/pg": "^8.23.1",
"oxfmt": "^0.56.0",
"oxlint": "^1.71.0",
"pino-pretty": "^13.1.3",
@@ -521,6 +525,110 @@
"node": ">=12"
}
},
"node_modules/@electric-sql/pglite": {
"version": "0.5.5",
"resolved": "https://registry.npmjs.org/@electric-sql/pglite/-/pglite-0.5.5.tgz",
"integrity": "sha512-QeZ+oB7QaU4EJj2awrB8hwFZ5TXxLRBQ9Gfq0dQauYDdWsRtuWRCERgtri35vRfL07KQHlVlc/4Qxvn7a5AXeA==",
"dev": true,
"license": "Apache-2.0"
},
"node_modules/@electric-sql/pglite-age": {
"version": "0.0.6",
"resolved": "https://registry.npmjs.org/@electric-sql/pglite-age/-/pglite-age-0.0.6.tgz",
"integrity": "sha512-cnj+KxJUgx6WdwFgIbu2wQSu5NKNOf1TNfLAwVTHZZGkTTyA8FW8FrFZ06jKUkKoW7Fad8X0O4FPb/I+u5n5Rw==",
"dev": true,
"license": "Apache-2.0",
"peer": true,
"peerDependencies": {
"@electric-sql/pglite": "0.5.5"
}
},
"node_modules/@electric-sql/pglite-pg_hashids": {
"version": "0.0.6",
"resolved": "https://registry.npmjs.org/@electric-sql/pglite-pg_hashids/-/pglite-pg_hashids-0.0.6.tgz",
"integrity": "sha512-723NvdqXz70rTjC5OKQHmmNNogTcZxPLAJf0NRDanz3zz77nBOCSlTL/UKkah90UVk5kqnbrwIcHqDVGmmONDA==",
"dev": true,
"license": "Apache-2.0",
"peer": true,
"peerDependencies": {
"@electric-sql/pglite": "0.5.5"
}
},
"node_modules/@electric-sql/pglite-pg_ivm": {
"version": "0.0.6",
"resolved": "https://registry.npmjs.org/@electric-sql/pglite-pg_ivm/-/pglite-pg_ivm-0.0.6.tgz",
"integrity": "sha512-LroA6kGyIEeB1/P7RbAaDHDCgu6+0scXzNe4ahN73f/PfexVN4h1k7HXOUS2su7tAHVMOBRlfenaWl3J/KhSmg==",
"dev": true,
"license": "Apache-2.0",
"peer": true,
"peerDependencies": {
"@electric-sql/pglite": "0.5.5"
}
},
"node_modules/@electric-sql/pglite-pg_textsearch": {
"version": "0.0.7",
"resolved": "https://registry.npmjs.org/@electric-sql/pglite-pg_textsearch/-/pglite-pg_textsearch-0.0.7.tgz",
"integrity": "sha512-gBWaCxBcFPrmecgnycTXVJrAM39qEsXqZEuQZg1itlnIti1233htFYkOg9xp+FFE8R0Z7XmjbSJxjFoG60LD7w==",
"dev": true,
"license": "Apache-2.0",
"peer": true,
"peerDependencies": {
"@electric-sql/pglite": "0.5.5"
}
},
"node_modules/@electric-sql/pglite-pg_uuidv7": {
"version": "0.0.6",
"resolved": "https://registry.npmjs.org/@electric-sql/pglite-pg_uuidv7/-/pglite-pg_uuidv7-0.0.6.tgz",
"integrity": "sha512-ffQARR67C9gFIhoaqSF/J0IUM6STJASrmcRvhlOokv414WBkl+wiNHBslw7hc3tYSZmg41Y7L1tjxNPNZn1wPA==",
"dev": true,
"license": "Apache-2.0",
"peer": true,
"peerDependencies": {
"@electric-sql/pglite": "0.5.5"
}
},
"node_modules/@electric-sql/pglite-pgtap": {
"version": "0.0.6",
"resolved": "https://registry.npmjs.org/@electric-sql/pglite-pgtap/-/pglite-pgtap-0.0.6.tgz",
"integrity": "sha512-2Xn9xevILlNHTAeLZJnrB7DtOoB2qEiXm+tPaP0cm4n1a9zsAb4HQBRgsQhWKF7PJn9qMGJYFMuWWoMaxLTXwA==",
"dev": true,
"license": "Apache-2.0",
"peer": true,
"peerDependencies": {
"@electric-sql/pglite": "0.5.5"
}
},
"node_modules/@electric-sql/pglite-pgvector": {
"version": "0.0.6",
"resolved": "https://registry.npmjs.org/@electric-sql/pglite-pgvector/-/pglite-pgvector-0.0.6.tgz",
"integrity": "sha512-KiPeiS/BMmjO71igzgexJAu5aPb3Y2yUjNTSMdti4ZL3zlqZFl4eLgbzLBGyG5Cxr9xFwYFmHh61ZTq6w6hUJA==",
"dev": true,
"license": "Apache-2.0",
"peer": true,
"peerDependencies": {
"@electric-sql/pglite": "0.5.5"
}
},
"node_modules/@electric-sql/pglite-socket": {
"version": "0.2.8",
"resolved": "https://registry.npmjs.org/@electric-sql/pglite-socket/-/pglite-socket-0.2.8.tgz",
"integrity": "sha512-s/hDC799oT1YLrQoyfiBKqwFcIn0m6OUEuU2ENdzr3uxW5rJBM73hpQeiBmSs+pqDD2A3PH4a8839ExqoROVmQ==",
"dev": true,
"license": "Apache-2.0",
"bin": {
"pglite-server": "dist/scripts/server.js"
},
"peerDependencies": {
"@electric-sql/pglite": "0.5.5",
"@electric-sql/pglite-age": "0.0.6",
"@electric-sql/pglite-pg_hashids": "0.0.6",
"@electric-sql/pglite-pg_ivm": "0.0.6",
"@electric-sql/pglite-pg_textsearch": "0.0.7",
"@electric-sql/pglite-pg_uuidv7": "0.0.6",
"@electric-sql/pglite-pgtap": "0.0.6",
"@electric-sql/pglite-pgvector": "0.0.6"
}
},
"node_modules/@embra/i18n": {
"version": "0.0.10",
"resolved": "https://registry.npmjs.org/@embra/i18n/-/i18n-0.0.10.tgz",
@@ -4298,6 +4406,18 @@
"@types/node": "*"
}
},
"node_modules/@types/pg": {
"version": "8.23.1",
"resolved": "https://registry.npmjs.org/@types/pg/-/pg-8.23.1.tgz",
"integrity": "sha512-fKVHpikPdg4GKks3JuLEhvwSyvwzF23hnabPy6DD8ljVbC7+6J5dQzdv4arV6jqq57djnMgs1HKBxX4P8aBI3A==",
"dev": true,
"license": "MIT",
"dependencies": {
"@types/node": "*",
"pg-protocol": "*",
"pg-types": "^2.2.0"
}
},
"node_modules/@types/react": {
"version": "19.2.17",
"resolved": "https://registry.npmjs.org/@types/react/-/react-19.2.17.tgz",
@@ -8231,6 +8351,95 @@
"url": "https://github.com/sponsors/KillyMXI"
}
},
"node_modules/pg": {
"version": "8.23.0",
"resolved": "https://registry.npmjs.org/pg/-/pg-8.23.0.tgz",
"integrity": "sha512-Ip2EQCngowJLGOfCwkFhPXU7/ljlhn6Rxlmy4XYfL2Y+vyRM59+8uR2xqRWKdYmbXmxCFOAmKxBuSUCdF34qLg==",
"license": "MIT",
"dependencies": {
"pg-connection-string": "^2.14.0",
"pg-pool": "^3.14.0",
"pg-protocol": "^1.16.0",
"pg-types": "2.2.0",
"pgpass": "1.0.5"
},
"engines": {
"node": ">= 16.0.0"
},
"optionalDependencies": {
"pg-cloudflare": "^1.4.0"
},
"peerDependencies": {
"pg-native": ">=3.0.1"
},
"peerDependenciesMeta": {
"pg-native": {
"optional": true
}
}
},
"node_modules/pg-cloudflare": {
"version": "1.4.0",
"resolved": "https://registry.npmjs.org/pg-cloudflare/-/pg-cloudflare-1.4.0.tgz",
"integrity": "sha512-Vo7z/6rrQYxpNRylp4Tlob2elzbh+N/MOQbxFVWCxS7oEx6jF53GTJFxK2WWpKuBRkmiin4Mt+xofFDjx09R0A==",
"license": "MIT",
"optional": true
},
"node_modules/pg-connection-string": {
"version": "2.14.0",
"resolved": "https://registry.npmjs.org/pg-connection-string/-/pg-connection-string-2.14.0.tgz",
"integrity": "sha512-XwWDGcLRGCXAR8F/AM5bG7Q+A3Wm2s6QeEjlOKZLlH3UYcguiqCWKyWXVag5TLTIjR7oOJUY8kcADaZgWPyLeg==",
"license": "MIT"
},
"node_modules/pg-int8": {
"version": "1.0.1",
"resolved": "https://registry.npmjs.org/pg-int8/-/pg-int8-1.0.1.tgz",
"integrity": "sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw==",
"license": "ISC",
"engines": {
"node": ">=4.0.0"
}
},
"node_modules/pg-pool": {
"version": "3.14.0",
"resolved": "https://registry.npmjs.org/pg-pool/-/pg-pool-3.14.0.tgz",
"integrity": "sha512-gKtPkFdQPU3DksooVLi9LsjZxrsBUZIpa+7aVx+LV5pNh0KzP4Zleud2po+ConrxbuXGBJ6Hfer6hdgpIBpBaw==",
"license": "MIT",
"peerDependencies": {
"pg": ">=8.0"
}
},
"node_modules/pg-protocol": {
"version": "1.16.0",
"resolved": "https://registry.npmjs.org/pg-protocol/-/pg-protocol-1.16.0.tgz",
"integrity": "sha512-sILXutLVjCLjcDuOmvhX5e2Z4cS5qG/6Bu3VkpFwdf/633ElGLpEh9bgmuI5I4sqKqkifQiGyiCcx1HdtrK7tg==",
"license": "MIT"
},
"node_modules/pg-types": {
"version": "2.2.0",
"resolved": "https://registry.npmjs.org/pg-types/-/pg-types-2.2.0.tgz",
"integrity": "sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA==",
"license": "MIT",
"dependencies": {
"pg-int8": "1.0.1",
"postgres-array": "~2.0.0",
"postgres-bytea": "~1.0.0",
"postgres-date": "~1.0.4",
"postgres-interval": "^1.1.0"
},
"engines": {
"node": ">=4"
}
},
"node_modules/pgpass": {
"version": "1.0.5",
"resolved": "https://registry.npmjs.org/pgpass/-/pgpass-1.0.5.tgz",
"integrity": "sha512-FdW9r/jQZhSeohs1Z3sI1yxFQNFvMcnmfuj4WBMUTxOrAyLMaTcE1aAMBiTlbMNaXvBCQuVi0R7hd8udDSP7ug==",
"license": "MIT",
"dependencies": {
"split2": "^4.1.0"
}
},
"node_modules/picocolors": {
"version": "1.1.1",
"resolved": "https://registry.npmjs.org/picocolors/-/picocolors-1.1.1.tgz",
@@ -8354,6 +8563,45 @@
"node": "^10 || ^12 || >=14"
}
},
"node_modules/postgres-array": {
"version": "2.0.0",
"resolved": "https://registry.npmjs.org/postgres-array/-/postgres-array-2.0.0.tgz",
"integrity": "sha512-VpZrUqU5A69eQyW2c5CA1jtLecCsN2U/bD6VilrFDWq5+5UIEVO7nazS3TEcHf1zuPYO/sqGvUvW62g86RXZuA==",
"license": "MIT",
"engines": {
"node": ">=4"
}
},
"node_modules/postgres-bytea": {
"version": "1.0.1",
"resolved": "https://registry.npmjs.org/postgres-bytea/-/postgres-bytea-1.0.1.tgz",
"integrity": "sha512-5+5HqXnsZPE65IJZSMkZtURARZelel2oXUEO8rH83VS/hxH5vv1uHquPg5wZs8yMAfdv971IU+kcPUczi7NVBQ==",
"license": "MIT",
"engines": {
"node": ">=0.10.0"
}
},
"node_modules/postgres-date": {
"version": "1.0.7",
"resolved": "https://registry.npmjs.org/postgres-date/-/postgres-date-1.0.7.tgz",
"integrity": "sha512-suDmjLVQg78nMK2UZ454hAG+OAW+HQPZ6n++TNDUX+L0+uUlLywnoxJKDou51Zm+zTCjrCl0Nq6J9C5hP9vK/Q==",
"license": "MIT",
"engines": {
"node": ">=0.10.0"
}
},
"node_modules/postgres-interval": {
"version": "1.2.0",
"resolved": "https://registry.npmjs.org/postgres-interval/-/postgres-interval-1.2.0.tgz",
"integrity": "sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ==",
"license": "MIT",
"dependencies": {
"xtend": "^4.0.0"
},
"engines": {
"node": ">=0.10.0"
}
},
"node_modules/process": {
"version": "0.11.10",
"resolved": "https://registry.npmjs.org/process/-/process-0.11.10.tgz",
+5
View File
@@ -21,6 +21,7 @@
"generate:catalog": "node scripts/generate-catalog.ts",
"export:feishu-permissions": "node scripts/export-feishu-permissions.ts",
"generate:provider": "node scripts/generate-provider.ts",
"runtime:migrate": "node scripts/runtime-data.ts migrate",
"runtime:data": "node scripts/runtime-data.ts",
"lint": "oxlint .",
"lint:fix": "oxlint . --fix",
@@ -52,11 +53,14 @@
"minisearch": "^7.2.0",
"mqtt": "^5.15.2",
"nodemailer": "^9.0.3",
"pg": "^8.23.0",
"pino": "^10.3.1",
"rss-parser": "^3.13.0",
"zod": "^4.4.3"
},
"devDependencies": {
"@electric-sql/pglite": "^0.5.5",
"@electric-sql/pglite-socket": "^0.2.8",
"@hyrious/configs": "^0.1.6",
"@types/ali-oss": "^6.23.3",
"@types/busboy": "^1.5.4",
@@ -64,6 +68,7 @@
"@types/mdast": "^4.0.4",
"@types/node": "^26.0.1",
"@types/nodemailer": "^8.0.1",
"@types/pg": "^8.23.1",
"oxfmt": "^0.56.0",
"oxlint": "^1.71.0",
"pino-pretty": "^13.1.3",
+69 -16
View File
@@ -1,8 +1,11 @@
import { mkdir } from "node:fs/promises";
import { join, resolve } from "node:path";
import { parseArgs } from "node:util";
import { Pool } from "pg";
import { logger } from "../src/server/logger.ts";
import { createSecretCodec } from "../src/server/secrets/secret-codec.ts";
import { SqliteRuntimeDatabase } from "../src/server/storage/sqlite-runtime-store.ts";
import { assertPostgresDatabaseUrl, createNodeRuntimeDatabase } from "../src/server/storage/node-runtime-database.ts";
import { migratePostgresDatabase } from "../src/server/storage/postgres-migrations.ts";
const { positionals, values: options } = parseArgs({
args: process.argv.slice(2),
@@ -16,12 +19,16 @@ const { positionals, values: options } = parseArgs({
});
const [command] = positionals;
if (positionals.length !== 1 || (command !== "reset" && command !== "rotate-key")) {
if (positionals.length !== 1 || (command !== "migrate" && command !== "reset" && command !== "rotate-key")) {
printUsageAndExit();
}
const nextEncryptionKey = process.env.OOMOL_CONNECT_NEW_ENCRYPTION_KEY;
if (command === "rotate-key") {
if (command === "migrate") {
if (options.plain || options.yes || options["data-dir"]) {
throw new Error("migrate does not accept --plain, --yes, or --data-dir.");
}
} else if (command === "rotate-key") {
if (options.yes) {
throw new Error("--yes is only valid with reset.");
}
@@ -37,31 +44,77 @@ if (command === "rotate-key") {
}
}
const databaseUrl = process.env.OOMOL_CONNECT_DATABASE_URL?.trim();
const dataDir = resolve(options["data-dir"] ?? process.env.OOMOL_CONNECT_DATA_DIR ?? join(process.cwd(), "data"));
const databasePath = join(dataDir, "connect.sqlite");
const secretCodec = createSecretCodec(process.env.OOMOL_CONNECT_ENCRYPTION_KEY);
await mkdir(dataDir, { recursive: true });
const database = new SqliteRuntimeDatabase(databasePath, { secretCodec });
try {
if (command === "rotate-key") {
await database.rotateSecretCodec(createSecretCodec(options.plain ? undefined : nextEncryptionKey));
console.log(`Rotated runtime secret encryption in ${databasePath}.`);
if (command === "migrate") {
if (!databaseUrl) {
console.log("SQLite migrations are applied automatically when the local runtime database opens.");
} else {
database.resetRuntimeData();
console.log(`Reset runtime data in ${databasePath}.`);
assertPostgresDatabaseUrl(databaseUrl);
const pool = new Pool({
application_name: "open-connector-migrate",
connectionString: databaseUrl,
connectionTimeoutMillis: readPositiveIntegerEnv("OOMOL_CONNECT_DATABASE_CONNECT_TIMEOUT_MS", 10_000),
max: 1,
});
try {
await migratePostgresDatabase({ pool, logger });
} finally {
await pool.end();
}
}
} else {
const secretCodec = createSecretCodec(process.env.OOMOL_CONNECT_ENCRYPTION_KEY);
if (!databaseUrl) {
await mkdir(dataDir, { recursive: true });
}
const database = databaseUrl
? await createNodeRuntimeDatabase({
backend: "postgresql",
connectionString: databaseUrl,
secretCodec,
poolMax: readPositiveIntegerEnv("OOMOL_CONNECT_DATABASE_POOL_MAX", 10),
connectionTimeoutMs: readPositiveIntegerEnv("OOMOL_CONNECT_DATABASE_CONNECT_TIMEOUT_MS", 10_000),
})
: await createNodeRuntimeDatabase({
backend: "sqlite",
path: databasePath,
secretCodec,
});
const target = databaseUrl ? "the PostgreSQL runtime database" : databasePath;
try {
if (command === "rotate-key") {
await database.rotateSecretCodec(createSecretCodec(options.plain ? undefined : nextEncryptionKey));
console.log(`Rotated runtime secret encryption in ${target}.`);
} else {
await database.resetRuntimeData();
console.log(`Reset runtime data in ${target}.`);
}
} finally {
await database.close();
}
} finally {
database.close();
}
function printUsageAndExit(): never {
console.error(`Usage:
node scripts/runtime-data.ts migrate
node scripts/runtime-data.ts reset --yes [--data-dir ./data]
node scripts/runtime-data.ts rotate-key [--data-dir ./data]
node scripts/runtime-data.ts rotate-key --plain [--data-dir ./data]
Set OOMOL_CONNECT_ENCRYPTION_KEY to read/write encrypted local credential records.
Set OOMOL_CONNECT_DATABASE_URL to migrate or maintain a PostgreSQL runtime database.
Set OOMOL_CONNECT_ENCRYPTION_KEY to read/write encrypted runtime credential records.
Set OOMOL_CONNECT_NEW_ENCRYPTION_KEY when rotating to a new encryption key.`);
process.exit(1);
}
function readPositiveIntegerEnv(name: string, fallback: number): number {
const value = process.env[name];
if (value === undefined) {
return fallback;
}
const parsed = Number(value);
return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback;
}
+135 -85
View File
@@ -1,4 +1,5 @@
import type { IStagedTransitFileService } from "./files/transit-file-store.ts";
import type { ServerType } from "@hono/node-server";
import { S3Client } from "@aws-sdk/client-s3";
import { serve } from "@hono/node-server";
@@ -22,8 +23,8 @@ import { S3TransitFileService } from "./files/s3-transit-files.ts";
import { TransitFileService } from "./files/transit-files.ts";
import { logger } from "./logger.ts";
import { createSecretCodec } from "./secrets/secret-codec.ts";
import { createNodeRuntimeDatabase } from "./storage/node-runtime-database.ts";
import { DEFAULT_RUN_LIMIT } from "./storage/runtime-store.ts";
import { SqliteRuntimeDatabase } from "./storage/sqlite-runtime-store.ts";
const port = Number(process.env.PORT ?? 3000);
const hostname = process.env.HOST ?? "127.0.0.1";
@@ -32,92 +33,141 @@ const dataDir = process.env.OOMOL_CONNECT_DATA_DIR ?? join(process.cwd(), "data"
const transitFileTtlSeconds = readPositiveIntegerEnv("OOMOL_CONNECT_TRANSIT_FILE_TTL_SECONDS", 86_400);
const transitFileMaxBytes = readPositiveIntegerEnv("OOMOL_CONNECT_TRANSIT_FILE_MAX_BYTES", 100 * 1024 * 1024);
const runLimit = readPositiveIntegerEnv("OOMOL_CONNECT_RUN_LIMIT", DEFAULT_RUN_LIMIT);
const secretCodec = createSecretCodec(process.env.OOMOL_CONNECT_ENCRYPTION_KEY);
const adminToken = process.env.OOMOL_CONNECT_ADMIN_TOKEN;
const runtimeToken = process.env.OOMOL_CONNECT_RUNTIME_TOKEN;
const verifyRuntimeJwt = createRuntimeJwtVerifier({
jwksUri: process.env.OOMOL_CONNECT_JWKS_URI,
issuer: process.env.OOMOL_CONNECT_JWT_ISSUER,
audience: process.env.OOMOL_CONNECT_JWT_AUDIENCE,
});
const actionPolicy = new ActionPolicyService({
allowedActions: parseActionPolicyList(process.env.OOMOL_CONNECT_ALLOWED_ACTIONS),
blockedActions: parseActionPolicyList(process.env.OOMOL_CONNECT_BLOCKED_ACTIONS),
allowedProxies: parseActionPolicyList(process.env.OOMOL_CONNECT_ALLOWED_PROXIES),
blockedProxies: parseActionPolicyList(process.env.OOMOL_CONNECT_BLOCKED_PROXIES),
});
const allowedCustomOAuth = parseActionPolicyList(process.env.OOMOL_CONNECT_ALLOWED_CUSTOM_OAUTH);
setPrivateNetworkAccessAllowed(parsePrivateNetworkAccessFlag(process.env.OOMOL_CONNECT_ALLOW_PRIVATE_NETWORK));
setEgressTrustedHosts(parseEgressTrustedHosts(process.env.OOMOL_CONNECT_EGRESS_TRUSTED_HOSTS));
const builtRoot = join(process.cwd(), "dist/web");
const staticRoot = await resolveStaticRoot(builtRoot);
await mkdir(dataDir, { recursive: true });
const catalog = await loadCatalog(undefined, {
executableServices: Object.keys(executorModules),
});
const providerLoader = new ProviderLoader(executorModules);
const runtimeDatabase = new SqliteRuntimeDatabase(join(dataDir, "connect.sqlite"), {
logger,
secretCodec,
runLimit,
});
const transitFiles = createTransitFileService();
const transitFileTempDir = join(dataDir, "tmp", "transit-files");
await transitFiles.cleanupExpired();
await cleanupStagedTransitFiles(transitFileTempDir, transitFileTtlSeconds * 1000);
const { app, runtimeAuthConfigured } = await createConnectApp({
catalog,
providerLoader,
runtimeDatabase,
transitFiles,
uploadTransitFile: createNodeTransitFileUpload({ transitFiles, tempDir: transitFileTempDir }),
publicOrigin,
secretCodec,
adminToken,
runtimeToken,
verifyRuntimeJwt,
actionPolicy,
allowedCustomOAuth,
registerStaticRoutes: (app) => registerStaticRoutes(app, staticRoot),
logger,
});
const databaseUrl = optionalEnv("OOMOL_CONNECT_DATABASE_URL");
const databasePoolMax = readPositiveIntegerEnv("OOMOL_CONNECT_DATABASE_POOL_MAX", 10);
const databaseConnectTimeoutMs = readPositiveIntegerEnv("OOMOL_CONNECT_DATABASE_CONNECT_TIMEOUT_MS", 10_000);
process.once("SIGINT", () => {
runtimeDatabase.close();
process.exit(0);
});
process.once("SIGTERM", () => {
runtimeDatabase.close();
process.exit(0);
});
try {
await main();
} catch (error) {
logger.error({ err: error }, "connect server failed");
process.exitCode = 1;
}
serve(
{
fetch: app.fetch,
port,
hostname,
},
(info) => {
logger.info({ url: `http://${hostname}:${info.port}` }, "connect server listening");
logger.info({ dataDir }, "runtime data directory");
if (!adminToken) {
logger.warn("local admin authentication is disabled; set OOMOL_CONNECT_ADMIN_TOKEN to require bearer tokens");
}
if (!runtimeAuthConfigured) {
logger.warn(
"runtime API authentication is disabled; create a runtime token in the web console, set OOMOL_CONNECT_RUNTIME_TOKEN, or configure JWT authentication",
);
}
if (!secretCodec.encrypted) {
logger.warn(
"local data encryption is disabled; set OOMOL_CONNECT_ENCRYPTION_KEY to encrypt stored credentials, OAuth client configuration, pending OAuth state, and completed idempotent action responses",
);
}
if (!staticRoot) {
logger.warn("web console assets are not built; use http://localhost:5173 for local console development");
}
},
);
async function main(): Promise<void> {
setPrivateNetworkAccessAllowed(parsePrivateNetworkAccessFlag(process.env.OOMOL_CONNECT_ALLOW_PRIVATE_NETWORK));
setEgressTrustedHosts(parseEgressTrustedHosts(process.env.OOMOL_CONNECT_EGRESS_TRUSTED_HOSTS));
const secretCodec = createSecretCodec(process.env.OOMOL_CONNECT_ENCRYPTION_KEY);
const adminToken = process.env.OOMOL_CONNECT_ADMIN_TOKEN;
const runtimeToken = process.env.OOMOL_CONNECT_RUNTIME_TOKEN;
const verifyRuntimeJwt = createRuntimeJwtVerifier({
jwksUri: process.env.OOMOL_CONNECT_JWKS_URI,
issuer: process.env.OOMOL_CONNECT_JWT_ISSUER,
audience: process.env.OOMOL_CONNECT_JWT_AUDIENCE,
});
const actionPolicy = new ActionPolicyService({
allowedActions: parseActionPolicyList(process.env.OOMOL_CONNECT_ALLOWED_ACTIONS),
blockedActions: parseActionPolicyList(process.env.OOMOL_CONNECT_BLOCKED_ACTIONS),
allowedProxies: parseActionPolicyList(process.env.OOMOL_CONNECT_ALLOWED_PROXIES),
blockedProxies: parseActionPolicyList(process.env.OOMOL_CONNECT_BLOCKED_PROXIES),
});
const allowedCustomOAuth = parseActionPolicyList(process.env.OOMOL_CONNECT_ALLOWED_CUSTOM_OAUTH);
await mkdir(dataDir, { recursive: true });
const staticRoot = await resolveStaticRoot(join(process.cwd(), "dist/web"));
const catalog = await loadCatalog(undefined, {
executableServices: Object.keys(executorModules),
});
const runtimeDatabase = databaseUrl
? await createNodeRuntimeDatabase({
backend: "postgresql",
connectionString: databaseUrl,
logger,
secretCodec,
runLimit,
poolMax: databasePoolMax,
connectionTimeoutMs: databaseConnectTimeoutMs,
})
: await createNodeRuntimeDatabase({
backend: "sqlite",
path: join(dataDir, "connect.sqlite"),
logger,
secretCodec,
runLimit,
});
try {
const transitFiles = createTransitFileService();
const transitFileTempDir = join(dataDir, "tmp", "transit-files");
await transitFiles.cleanupExpired();
await cleanupStagedTransitFiles(transitFileTempDir, transitFileTtlSeconds * 1000);
const { app, runtimeAuthConfigured } = await createConnectApp({
catalog,
providerLoader: new ProviderLoader(executorModules),
runtimeDatabase,
transitFiles,
uploadTransitFile: createNodeTransitFileUpload({ transitFiles, tempDir: transitFileTempDir }),
publicOrigin,
secretCodec,
adminToken,
runtimeToken,
verifyRuntimeJwt,
actionPolicy,
allowedCustomOAuth,
registerStaticRoutes: (app) => registerStaticRoutes(app, staticRoot),
logger,
});
const server = serve(
{
fetch: app.fetch,
port,
hostname,
},
(info) => {
logger.info({ url: `http://${hostname}:${info.port}` }, "connect server listening");
logger.info({ dataDir }, "runtime data directory");
logger.info({ backend: databaseUrl ? "postgresql" : "sqlite" }, "runtime database ready");
if (!adminToken) {
logger.warn("local admin authentication is disabled; set OOMOL_CONNECT_ADMIN_TOKEN to require bearer tokens");
}
if (!runtimeAuthConfigured) {
logger.warn(
"runtime API authentication is disabled; create a runtime token in the web console, set OOMOL_CONNECT_RUNTIME_TOKEN, or configure JWT authentication",
);
}
if (!secretCodec.encrypted) {
logger.warn(
"runtime data encryption is disabled; set OOMOL_CONNECT_ENCRYPTION_KEY to encrypt stored credentials, OAuth client configuration, pending OAuth state, and completed idempotent action responses",
);
}
if (!staticRoot) {
logger.warn("web console assets are not built; use http://localhost:5173 for local console development");
}
},
);
await waitForShutdown(server);
} finally {
await runtimeDatabase.close();
}
}
function waitForShutdown(server: ServerType): Promise<void> {
return new Promise((resolve, reject) => {
let closing = false;
const shutdown = (): void => {
if (closing) {
return;
}
closing = true;
server.close((error) => {
process.removeListener("SIGINT", shutdown);
process.removeListener("SIGTERM", shutdown);
if (error) {
reject(error);
} else {
resolve();
}
});
};
process.once("SIGINT", shutdown);
process.once("SIGTERM", shutdown);
});
}
async function resolveStaticRoot(root: string): Promise<string | undefined> {
try {
@@ -0,0 +1,54 @@
import type { RuntimeLogger } from "../../core/types.ts";
import type { ISecretCodec } from "../secrets/secret-codec-core.ts";
import type { RuntimeDatabase } from "./runtime-database.ts";
import { PostgresRuntimeDatabase } from "./postgres-runtime-store.ts";
import { SqliteRuntimeDatabase } from "./sqlite-runtime-store.ts";
export interface NodeRuntimeDatabase extends RuntimeDatabase {
close(): void | Promise<void>;
resetRuntimeData(): void | Promise<void>;
rotateSecretCodec(nextSecretCodec: ISecretCodec): Promise<void>;
}
interface CommonOptions {
logger?: RuntimeLogger;
runLimit?: number;
secretCodec?: ISecretCodec;
}
interface SqliteOptions extends CommonOptions {
backend: "sqlite";
path: string;
}
interface PostgresOptions extends CommonOptions {
backend: "postgresql";
connectionString: string;
poolMax?: number;
connectionTimeoutMs?: number;
}
export type NodeRuntimeDatabaseOptions = SqliteOptions | PostgresOptions;
export async function createNodeRuntimeDatabase(options: NodeRuntimeDatabaseOptions): Promise<NodeRuntimeDatabase> {
if (options.backend === "sqlite") {
return new SqliteRuntimeDatabase(options.path, options);
}
const connectionString = options.connectionString.trim();
assertPostgresDatabaseUrl(connectionString);
return await PostgresRuntimeDatabase.open(connectionString, options);
}
export function assertPostgresDatabaseUrl(value: string): void {
let protocol: string;
try {
protocol = new URL(value).protocol;
} catch {
throw new Error("OOMOL_CONNECT_DATABASE_URL must be a valid PostgreSQL connection URL.");
}
if (protocol !== "postgres:" && protocol !== "postgresql:") {
throw new Error("OOMOL_CONNECT_DATABASE_URL must use the postgres: or postgresql: scheme.");
}
}
@@ -0,0 +1,121 @@
import { randomUUID } from "node:crypto";
import { Pool } from "pg";
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest";
import { migratePostgresDatabase } from "./postgres-migrations.ts";
import { PostgresRuntimeDatabase } from "./postgres-runtime-store.ts";
const testPostgresUrl = process.env.TEST_POSTGRES_URL;
describe.skipIf(!testPostgresUrl)("PostgreSQL runtime integration", () => {
const schemas: string[] = [];
let adminPool: Pool;
let runtimeUrl: string;
let first: PostgresRuntimeDatabase;
let second: PostgresRuntimeDatabase;
beforeAll(async () => {
adminPool = new Pool({ connectionString: testPostgresUrl });
runtimeUrl = await createTestSchema("runtime");
const pool = new Pool({ connectionString: runtimeUrl, max: 1 });
try {
await migratePostgresDatabase({ pool });
} finally {
await pool.end();
}
});
beforeEach(async () => {
first = await PostgresRuntimeDatabase.open(runtimeUrl, { poolMax: 2 });
second = await PostgresRuntimeDatabase.open(runtimeUrl, { poolMax: 2 });
await first.resetRuntimeData();
});
afterEach(async () => {
await first.close();
await second.close();
});
afterAll(async () => {
for (const schema of schemas) {
await adminPool.query(`drop schema ${schema} cascade`);
}
await adminPool.end();
});
it("serializes concurrent migration runners across PostgreSQL sessions", async () => {
const url = await createTestSchema("migrations");
const firstPool = new Pool({ connectionString: url, max: 1 });
const secondPool = new Pool({ connectionString: url, max: 1 });
try {
await Promise.all([migratePostgresDatabase({ pool: firstPool }), migratePostgresDatabase({ pool: secondPool })]);
const result = await firstPool.query<{ name: string }>("select name from runtime_migrations order by name");
expect(result.rows.map((row) => row.name)).toEqual(
expect.arrayContaining(["0010_runtime.sql", "0011_runtime_token_connection_scope.sql"]),
);
} finally {
await firstPool.end();
await secondPool.end();
}
});
it("atomically consumes OAuth state across database instances", async () => {
await first.oauthStateStore.set({
service: "gmail",
state: "state-1",
createdAt: "2026-06-30T00:00:00.000Z",
});
const results = await Promise.all([first.oauthStateStore.take("state-1"), second.oauthStateStore.take("state-1")]);
expect(results.filter((result) => result !== undefined)).toHaveLength(1);
});
it("atomically claims idempotency keys across database instances", async () => {
const claim = {
keyHash: "key-hash",
requestHash: "request-hash",
now: "2026-06-30T00:00:00.000Z",
expiresAt: "2026-07-01T00:00:00.000Z",
};
const results = await Promise.all([
first.idempotencyStore.claim({ ...claim, claimId: "claim-1" }),
second.idempotencyStore.claim({ ...claim, claimId: "claim-2" }),
]);
expect(results.map((result) => result.kind).sort()).toEqual(["acquired", "in_progress"]);
});
it("rejects stale connection revisions across database instances", async () => {
const credential = {
authType: "api_key" as const,
apiKey: "github-token",
values: { apiKey: "github-token" },
profile: {
accountId: "github:octocat",
displayName: "octocat",
grantedScopes: [],
},
metadata: {},
};
const created = await first.connectionStore.set("github", "default", credential);
const updated = await second.connectionStore.set("github", "default", {
...credential,
apiKey: "updated-token",
});
await expect(
first.connectionStore.updateCredential({
...created,
credential: { ...credential, apiKey: "stale-token" },
}),
).resolves.toBe(false);
expect(updated.id).toBe(created.id);
});
async function createTestSchema(label: string): Promise<string> {
const schema = `open_connector_${label}_${randomUUID().replaceAll("-", "")}`;
await adminPool.query(`create schema ${schema}`);
schemas.push(schema);
const url = new URL(testPostgresUrl!);
url.searchParams.set("options", `-c search_path=${schema}`);
return url.toString();
}
});
+120
View File
@@ -0,0 +1,120 @@
import type { RuntimeLogger } from "../../core/types.ts";
import type { Pool, PoolClient } from "pg";
import { readFileSync, readdirSync } from "node:fs";
const migrationDirectory = new URL("../../../migrations/postgresql/", import.meta.url);
const migrationLockNamespace = 1_326_382_671;
const migrationLockId = 1;
interface PostgresMigration {
name: string;
sql: string;
}
export interface PostgresMigrationOptions {
pool: Pool;
logger?: RuntimeLogger;
}
export async function migratePostgresDatabase(options: PostgresMigrationOptions): Promise<void> {
const client = await options.pool.connect();
let destroyClient = false;
try {
await client.query("select pg_advisory_lock($1, $2)", [migrationLockNamespace, migrationLockId]);
await client.query(`
create table if not exists runtime_migrations (
name text primary key,
applied_at text not null
)
`);
const startedAt = Date.now();
const migrations = readPostgresMigrations();
const applied = await readAppliedMigrations(client);
let newlyAppliedCount = 0;
for (const migration of migrations) {
if (applied.has(migration.name)) {
continue;
}
const migrationStartedAt = Date.now();
options.logger?.info({ migration: migration.name }, "postgres migration started");
await client.query("begin");
try {
await client.query(migration.sql);
await client.query("insert into runtime_migrations (name, applied_at) values ($1, $2)", [
migration.name,
new Date().toISOString(),
]);
await client.query("commit");
} catch (error) {
await client.query("rollback");
options.logger?.error(
{ migration: migration.name, durationMs: Date.now() - migrationStartedAt, err: error },
"postgres migration failed",
);
throw error;
}
applied.add(migration.name);
newlyAppliedCount += 1;
options.logger?.info(
{ migration: migration.name, durationMs: Date.now() - migrationStartedAt },
"postgres migration completed",
);
}
options.logger?.info(
{
migrationCount: migrations.length,
appliedCount: migrations.filter((migration) => applied.has(migration.name)).length,
newlyAppliedCount,
durationMs: Date.now() - startedAt,
},
"postgres migrations ready",
);
} finally {
try {
await client.query("select pg_advisory_unlock($1, $2)", [migrationLockNamespace, migrationLockId]);
} catch {
destroyClient = true;
}
client.release(destroyClient);
}
}
export async function assertPostgresSchemaReady(pool: Pool): Promise<void> {
const migrations = readPostgresMigrations();
const relation = await pool.query<{ name: string | null }>("select to_regclass($1) as name", ["runtime_migrations"]);
if (!relation.rows[0]?.name) {
throw new Error(
"PostgreSQL runtime schema is not initialized. Run `npm run runtime:migrate` before starting the server.",
);
}
const applied = await readAppliedMigrations(pool);
const missing = migrations.filter((migration) => !applied.has(migration.name)).map((migration) => migration.name);
if (missing.length > 0) {
throw new Error(
`PostgreSQL runtime schema is not ready. Missing migrations: ${missing.join(", ")}. Run \`npm run runtime:migrate\` before starting the server.`,
);
}
}
function readPostgresMigrations(): PostgresMigration[] {
return readdirSync(migrationDirectory)
.filter((name) => /^\d+_.*\.sql$/.test(name))
.sort()
.map((name) => ({
name,
sql: readFileSync(new URL(name, migrationDirectory), "utf8"),
}));
}
async function readAppliedMigrations(queryable: Pool | PoolClient): Promise<Set<string>> {
const result = await queryable.query<{ name: string }>("select name from runtime_migrations");
return new Set(result.rows.map((row) => row.name));
}
@@ -0,0 +1,383 @@
import type { RuntimeActionHttpResult } from "../api/runtime-api.ts";
import { PGlite } from "@electric-sql/pglite";
import { PGLiteSocketServer } from "@electric-sql/pglite-socket";
import { Pool } from "pg";
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest";
import { AesGcmSecretCodec } from "../secrets/secret-codec.ts";
import { assertPostgresSchemaReady, migratePostgresDatabase } from "./postgres-migrations.ts";
import { PostgresRuntimeDatabase } from "./postgres-runtime-store.ts";
import { RuntimeTokenService } from "./runtime-token-service.ts";
const githubProfile = {
accountId: "github:octocat",
displayName: "octocat",
grantedScopes: [],
};
interface PGliteTestServer {
database: PGlite;
server: PGLiteSocketServer;
url: string;
}
describe("PostgreSQL migrations with PGlite", () => {
let testServer: PGliteTestServer;
beforeAll(async () => {
testServer = await startPGliteTestServer();
});
afterAll(async () => {
await testServer.server.stop();
await testServer.database.close();
});
it("requires explicit migrations and recognizes the current schema", async () => {
await expect(PostgresRuntimeDatabase.open(testServer.url)).rejects.toThrow(
"Run `npm run runtime:migrate` before starting the server",
);
const pool = new Pool({ connectionString: testServer.url, max: 1 });
try {
await migratePostgresDatabase({ pool });
await expect(assertPostgresSchemaReady(pool)).resolves.toBeUndefined();
await expect(migratePostgresDatabase({ pool })).resolves.toBeUndefined();
await expect(pool.query("select name from runtime_migrations order by name")).resolves.toMatchObject({
rows: [{ name: "0010_runtime.sql" }, { name: "0011_runtime_token_connection_scope.sql" }],
});
await pool.query("delete from runtime_migrations where name = $1", ["0010_runtime.sql"]);
await expect(assertPostgresSchemaReady(pool)).rejects.toThrow("Missing migrations: 0010_runtime.sql");
await pool.query("insert into runtime_migrations (name, applied_at) values ($1, $2)", [
"0010_runtime.sql",
new Date().toISOString(),
]);
await pool.query("insert into runtime_migrations (name, applied_at) values ($1, $2)", [
"9999_future.sql",
new Date().toISOString(),
]);
await expect(assertPostgresSchemaReady(pool)).resolves.toBeUndefined();
} finally {
await pool.end();
}
});
});
describe("PostgresRuntimeDatabase with PGlite", () => {
let testServer: PGliteTestServer;
let database: PostgresRuntimeDatabase;
beforeAll(async () => {
testServer = await startPGliteTestServer();
const pool = new Pool({ connectionString: testServer.url, max: 1 });
try {
await migratePostgresDatabase({ pool });
} finally {
await pool.end();
}
});
beforeEach(async () => {
database = await PostgresRuntimeDatabase.open(testServer.url, { runLimit: 5 });
await database.resetRuntimeData();
});
afterEach(async () => {
await database.close();
});
afterAll(async () => {
await testServer.server.stop();
await testServer.database.close();
});
it("persists connections and OAuth data across database instances", async () => {
const connection = await database.connectionStore.set("github", "default", githubCredential("github-token"));
await database.oauthClientConfigStore.set({
service: "gmail",
clientId: "client-id",
clientSecret: "client-secret",
requestedScopes: ["gmail.readonly"],
extra: { tenant: "default" },
secretExtra: {},
});
await database.oauthStateStore.set({
service: "gmail",
state: "state-1",
createdAt: "2026-06-30T00:00:00.000Z",
});
await database.close();
database = await PostgresRuntimeDatabase.open(testServer.url);
await expect(database.connectionStore.get("github", "default")).resolves.toMatchObject({
id: connection.id,
credential: { apiKey: "github-token" },
});
await expect(database.oauthClientConfigStore.get("gmail")).resolves.toMatchObject({
clientId: "client-id",
clientSecret: "client-secret",
requestedScopes: ["gmail.readonly"],
});
await expect(database.oauthStateStore.take("state-1")).resolves.toMatchObject({ state: "state-1" });
await expect(database.oauthStateStore.take("state-1")).resolves.toBeUndefined();
});
it("preserves connection identity and rejects stale revisions", async () => {
const created = await database.connectionStore.set("github", "default", githubCredential("github-token"));
const updated = await database.connectionStore.set("github", "default", githubCredential("updated-token"));
expect(updated.id).toBe(created.id);
expect(updated.revision).not.toBe(created.revision);
await expect(
database.connectionStore.updateCredential({
...created,
credential: githubCredential("stale-token"),
}),
).resolves.toBe(false);
await expect(
database.connectionStore.updateCredential({
...updated,
credential: githubCredential("refreshed-token"),
}),
).resolves.toBe(true);
await expect(
database.connectionStore.updateCredential({
...updated,
credential: githubCredential("second-stale-token"),
}),
).resolves.toBe(false);
await expect(database.connectionStore.get("github", "default")).resolves.toMatchObject({
credential: { apiKey: "refreshed-token" },
});
});
it("claims, completes, replays, and expires idempotency records", async () => {
const first = {
keyHash: "key-hash",
requestHash: "request-hash",
claimId: "claim-1",
now: "2026-06-30T00:00:00.000Z",
expiresAt: "2026-06-30T01:00:00.000Z",
};
const second = {
...first,
claimId: "claim-2",
now: first.expiresAt,
expiresAt: "2026-06-30T02:00:00.000Z",
};
const response = successResponse({ claim: "current" });
await expect(database.idempotencyStore.claim(first)).resolves.toEqual({ kind: "acquired" });
await expect(database.idempotencyStore.claim({ ...first, claimId: "duplicate" })).resolves.toEqual({
kind: "in_progress",
});
await expect(
database.idempotencyStore.claim({ ...first, claimId: "conflict", requestHash: "different" }),
).resolves.toEqual({ kind: "conflict" });
await expect(database.idempotencyStore.claim(second)).resolves.toEqual({ kind: "acquired" });
await expect(
database.idempotencyStore.complete({
...first,
response: successResponse({ claim: "stale" }),
}),
).resolves.toBe(false);
await expect(database.idempotencyStore.complete({ ...second, response })).resolves.toBe(true);
await expect(
database.idempotencyStore.claim({
...second,
claimId: "claim-3",
now: "2026-06-30T01:30:00.000Z",
}),
).resolves.toEqual({ kind: "completed", response });
});
it("stores tokens and policy and filters paginated run logs", async () => {
const tokens = new RuntimeTokenService(database.runtimeTokenStore);
const token = await tokens.createToken("Claude Desktop", {
allowedActions: ["github.*"],
blockedActions: ["github.delete_repository"],
allowedProxies: ["github"],
allowedConnections: ["connection-work", "connection-personal"],
});
await expect(tokens.verifyToken(token.token)).resolves.toBe(true);
await expect(tokens.resolveToken(token.token)).resolves.toMatchObject({
allowedConnections: ["connection-work", "connection-personal"],
});
await expect(tokens.listTokens()).resolves.toMatchObject([
{
id: token.record.id,
allowedActions: ["github.*"],
blockedActions: ["github.delete_repository"],
allowedProxies: ["github"],
allowedConnections: ["connection-work", "connection-personal"],
},
]);
await expect(
tokens.updateTokenPolicy(token.record.id, {
allowedActions: ["github.get_current_user"],
blockedActions: [],
allowedProxies: [],
allowedConnections: ["connection-work"],
}),
).resolves.toMatchObject({
allowedActions: ["github.get_current_user"],
allowedConnections: ["connection-work"],
});
await expect(tokens.resolveToken(token.token)).resolves.toMatchObject({
allowedConnections: ["connection-work"],
});
const policy = {
rules: {
allowedActions: ["github.*"],
blockedActions: [],
allowedProxies: ["github"],
blockedProxies: [],
},
updatedAt: "2026-07-20T00:00:00.000Z",
};
await database.runtimePolicyStore.set(policy);
await expect(database.runtimePolicyStore.get()).resolves.toEqual(policy);
await database.runLogStore.add(createRun("gmail-1", "2026-06-30T00:00:00.000Z", "gmail.list", "gmail"));
await database.runLogStore.add(createRun("news-1", "2026-06-30T00:00:01.000Z"));
await database.runLogStore.add(createRun("gmail-2", "2026-06-30T00:00:02.000Z", "gmail.send", "gmail"));
const first = await database.runLogStore.list({ service: "gmail", limit: 1 });
expect(first.items.map((run) => run.id)).toEqual(["gmail-2"]);
expect(first.nextCursor).toBeTruthy();
await expect(
database.runLogStore.list({ service: "gmail", limit: 1, cursor: first.nextCursor }),
).resolves.toMatchObject({ items: [{ id: "gmail-1" }] });
await expect(database.runLogStore.get("news-1")).resolves.toMatchObject({ id: "news-1" });
await expect(tokens.revokeToken(token.record.id)).resolves.toBe(true);
await expect(tokens.verifyToken(token.token)).resolves.toBe(false);
});
it("keeps only the configured number of recent runs", async () => {
await database.close();
database = await PostgresRuntimeDatabase.open(testServer.url, { runLimit: 2 });
await database.runLogStore.add(createRun("run-1", "2026-06-30T00:00:00.000Z"));
await database.runLogStore.add(createRun("run-2", "2026-06-30T00:00:01.000Z"));
await database.runLogStore.add(createRun("run-3", "2026-06-30T00:00:02.000Z"));
await expect(database.runLogStore.list()).resolves.toMatchObject({
items: [{ id: "run-3" }, { id: "run-2" }],
});
});
it("rotates encrypted values and resets runtime data without removing migrations", async () => {
await database.close();
database = await PostgresRuntimeDatabase.open(testServer.url, {
secretCodec: new AesGcmSecretCodec("old-key"),
});
await database.connectionStore.set("github", "default", githubCredential("github-token"));
await database.oauthClientConfigStore.set({
service: "gmail",
clientId: "client-id",
clientSecret: "client-secret",
extra: {},
secretExtra: {},
});
await database.oauthStateStore.set({
service: "gmail",
state: "state-rotation",
createdAt: "2026-06-30T00:00:00.000Z",
});
const claim = {
keyHash: "key-hash",
requestHash: "request-hash",
claimId: "claim-1",
now: "2026-06-30T00:00:00.000Z",
expiresAt: "2026-07-01T00:00:00.000Z",
};
await database.idempotencyStore.claim(claim);
await database.idempotencyStore.complete({ ...claim, response: successResponse({ secret: "response" }) });
await database.rotateSecretCodec(new AesGcmSecretCodec("new-key"));
await database.close();
const withOldKey = await PostgresRuntimeDatabase.open(testServer.url, {
secretCodec: new AesGcmSecretCodec("old-key"),
});
await expect(withOldKey.connectionStore.get("github", "default")).rejects.toThrow();
await withOldKey.close();
database = await PostgresRuntimeDatabase.open(testServer.url, {
secretCodec: new AesGcmSecretCodec("new-key"),
});
await expect(database.connectionStore.get("github", "default")).resolves.toMatchObject({
credential: { apiKey: "github-token" },
});
await expect(database.oauthClientConfigStore.get("gmail")).resolves.toMatchObject({
clientSecret: "client-secret",
});
await expect(database.oauthStateStore.take("state-rotation")).resolves.toMatchObject({
state: "state-rotation",
});
await expect(database.idempotencyStore.claim({ ...claim, claimId: "claim-2" })).resolves.toEqual({
kind: "completed",
response: successResponse({ secret: "response" }),
});
await database.resetRuntimeData();
await expect(database.connectionStore.list()).resolves.toEqual([]);
const pool = new Pool({ connectionString: testServer.url, max: 1 });
try {
await expect(assertPostgresSchemaReady(pool)).resolves.toBeUndefined();
} finally {
await pool.end();
}
});
});
async function startPGliteTestServer(): Promise<PGliteTestServer> {
const database = await PGlite.create();
const server = new PGLiteSocketServer({
db: database,
host: "127.0.0.1",
port: 0,
maxConnections: 20,
});
await server.start();
const port = server.getServerConn().split(":").at(-1)!;
return {
database,
server,
url: `postgresql://postgres:[email protected]:${port}/postgres?sslmode=disable`,
};
}
function githubCredential(apiKey: string) {
return {
authType: "api_key" as const,
apiKey,
values: { apiKey },
profile: githubProfile,
metadata: {},
};
}
function createRun(id: string, startedAt: string, actionId = "hackernews.get_top_stories", service = "hackernews") {
return {
id,
service,
actionId,
caller: "http" as const,
startedAt,
completedAt: startedAt,
durationMs: 0,
ok: true,
};
}
function successResponse(data: unknown): RuntimeActionHttpResult {
return {
status: 200,
body: {
success: true,
message: "OK",
data,
meta: {},
},
};
}
@@ -0,0 +1,683 @@
import type { IConnectionStore, StoredConnection } from "../../connection-service.ts";
import type { TokenPolicy } from "../../core/action-policy.ts";
import type { ResolvedCredential, RuntimeLogger } from "../../core/types.ts";
import type { IOAuthClientConfigStore, OAuthClientConfig } from "../../oauth/oauth-client-config-service.ts";
import type { IOAuthStateStore, OAuthAuthorizationState } from "../../oauth/oauth-flow-service.ts";
import type { ISecretCodec } from "../secrets/secret-codec-core.ts";
import type {
CompleteIdempotencyInput,
IdempotencyClaimInput,
IdempotencyClaimResult,
IIdempotencyStore,
} from "./idempotency-store.ts";
import type { RuntimeDatabase } from "./runtime-database.ts";
import type { IRuntimePolicyStore, RuntimePolicyRecord } from "./runtime-policy-store.ts";
import type { IRunLogStore, RunLog, RunLogListInput, RunLogPage, RunLogWriteResult } from "./runtime-store.ts";
import type { IRuntimeTokenStore, RuntimeTokenRecord } from "./runtime-token-service.ts";
import type { PoolClient } from "pg";
import { Pool } from "pg";
import { parseRuntimeActionHttpResult } from "../api/runtime-api.ts";
import { PlainTextSecretCodec } from "../secrets/secret-codec-core.ts";
import { assertPostgresSchemaReady } from "./postgres-migrations.ts";
import { DEFAULT_RUN_LIMIT, decodeRunLogCursor, encodeRunLogCursor } from "./runtime-store.ts";
type RuntimeRow = Record<string, unknown>;
export interface PostgresRuntimeDatabaseOptions {
logger?: RuntimeLogger;
runLimit?: number;
secretCodec?: ISecretCodec;
poolMax?: number;
connectionTimeoutMs?: number;
}
export class PostgresRuntimeDatabase implements RuntimeDatabase {
readonly connectionStore: IConnectionStore;
readonly oauthClientConfigStore: IOAuthClientConfigStore;
readonly oauthStateStore: IOAuthStateStore;
readonly runtimeTokenStore: IRuntimeTokenStore;
readonly runtimePolicyStore: IRuntimePolicyStore;
readonly runLogStore: IRunLogStore;
readonly idempotencyStore: IIdempotencyStore;
private readonly pool: Pool;
private readonly secretCodec: ISecretCodec;
private constructor(pool: Pool, options: PostgresRuntimeDatabaseOptions) {
this.pool = pool;
this.secretCodec = options.secretCodec ?? new PlainTextSecretCodec();
this.connectionStore = new PostgresConnectionStore(pool, this.secretCodec);
this.oauthClientConfigStore = new PostgresOAuthClientConfigStore(pool, this.secretCodec);
this.oauthStateStore = new PostgresOAuthStateStore(pool, this.secretCodec);
this.runtimeTokenStore = new PostgresRuntimeTokenStore(pool);
this.runtimePolicyStore = new PostgresRuntimePolicyStore(pool);
this.runLogStore = new PostgresRunLogStore(pool, options.runLimit ?? DEFAULT_RUN_LIMIT);
this.idempotencyStore = new PostgresIdempotencyStore(pool, this.secretCodec);
}
static async open(
connectionString: string,
options: PostgresRuntimeDatabaseOptions = {},
): Promise<PostgresRuntimeDatabase> {
const pool = new Pool({
application_name: "open-connector",
connectionString,
connectionTimeoutMillis: options.connectionTimeoutMs ?? 10_000,
max: options.poolMax ?? 10,
});
pool.on("error", (error) => {
options.logger?.error({ err: error }, "postgres idle client error");
});
try {
await assertPostgresSchemaReady(pool);
return new PostgresRuntimeDatabase(pool, options);
} catch (error) {
await pool.end();
throw error;
}
}
async close(): Promise<void> {
await this.pool.end();
}
async resetRuntimeData(): Promise<void> {
await runInTransaction(this.pool, async (client) => {
await client.query(`
delete from connections;
delete from oauth_client_configs;
delete from oauth_states;
delete from runtime_tokens;
delete from runtime_policy;
delete from runs;
delete from idempotency_records;
`);
});
}
async rotateSecretCodec(nextSecretCodec: ISecretCodec): Promise<void> {
await runInTransaction(this.pool, async (client) => {
await client.query(
"lock table connections, oauth_client_configs, oauth_states, idempotency_records in access exclusive mode",
);
const connectionRows = await client.query<RuntimeRow>("select service, connection_name, value from connections");
const connections = await Promise.all(
connectionRows.rows.map(async (row) => ({
service: readString(row, "service"),
connectionName: readString(row, "connection_name"),
value: await nextSecretCodec.encode(await this.secretCodec.decode(readString(row, "value"))),
})),
);
for (const connection of connections) {
await client.query("update connections set value = $1 where service = $2 and connection_name = $3", [
connection.value,
connection.service,
connection.connectionName,
]);
}
const configRows = await client.query<RuntimeRow>("select service, value from oauth_client_configs");
const configs = await Promise.all(
configRows.rows.map(async (row) => ({
service: readString(row, "service"),
value: await nextSecretCodec.encode(await this.secretCodec.decode(readString(row, "value"))),
})),
);
for (const config of configs) {
await client.query("update oauth_client_configs set value = $1 where service = $2", [
config.value,
config.service,
]);
}
const stateRows = await client.query<RuntimeRow>("select state, value from oauth_states");
const states = await Promise.all(
stateRows.rows.map(async (row) => ({
state: readString(row, "state"),
value: await nextSecretCodec.encode(await this.secretCodec.decode(readString(row, "value"))),
})),
);
for (const state of states) {
await client.query("update oauth_states set value = $1 where state = $2", [state.value, state.state]);
}
const responseRows = await client.query<RuntimeRow>(
"select key_hash, response_value from idempotency_records where response_value is not null",
);
const responses = await Promise.all(
responseRows.rows.map(async (row) => ({
keyHash: readString(row, "key_hash"),
value: await nextSecretCodec.encode(await this.secretCodec.decode(readString(row, "response_value"))),
})),
);
for (const response of responses) {
await client.query("update idempotency_records set response_value = $1 where key_hash = $2", [
response.value,
response.keyHash,
]);
}
});
}
}
class PostgresConnectionStore implements IConnectionStore {
private readonly pool: Pool;
private readonly secretCodec: ISecretCodec;
constructor(pool: Pool, secretCodec: ISecretCodec) {
this.pool = pool;
this.secretCodec = secretCodec;
}
async get(service: string, connectionName: string): Promise<StoredConnection | undefined> {
const result = await this.pool.query<RuntimeRow>(
"select id, revision, value from connections where service = $1 and connection_name = $2",
[service, connectionName],
);
const row = result.rows[0];
return row
? {
id: readString(row, "id"),
revision: readString(row, "revision"),
service,
connectionName,
credential: parseJson<ResolvedCredential>(await this.secretCodec.decode(readString(row, "value"))),
}
: undefined;
}
async set(service: string, connectionName: string, credential: ResolvedCredential): Promise<StoredConnection> {
const result = await this.pool.query<RuntimeRow>(
`
insert into connections (id, revision, service, connection_name, value, updated_at)
values ($1, $2, $3, $4, $5, $6)
on conflict(service, connection_name) do update set
revision = excluded.revision,
value = excluded.value,
updated_at = excluded.updated_at
returning id, revision
`,
[
crypto.randomUUID(),
crypto.randomUUID(),
service,
connectionName,
await this.secretCodec.encode(JSON.stringify(credential)),
new Date().toISOString(),
],
);
const row = result.rows[0]!;
return {
id: readString(row, "id"),
revision: readString(row, "revision"),
service,
connectionName,
credential,
};
}
async updateCredential(input: StoredConnection): Promise<boolean> {
const result = await this.pool.query(
`
update connections
set revision = $1, value = $2, updated_at = $3
where service = $4 and connection_name = $5 and id = $6 and revision = $7
returning id
`,
[
crypto.randomUUID(),
await this.secretCodec.encode(JSON.stringify(input.credential)),
new Date().toISOString(),
input.service,
input.connectionName,
input.id,
input.revision,
],
);
return (result.rowCount ?? 0) > 0;
}
async delete(service: string, connectionName: string): Promise<void> {
await this.pool.query("delete from connections where service = $1 and connection_name = $2", [
service,
connectionName,
]);
}
async list(): Promise<StoredConnection[]> {
const result = await this.pool.query<RuntimeRow>(
"select id, revision, service, connection_name, value from connections order by service, connection_name",
);
return await Promise.all(
result.rows.map(async (row) => ({
id: readString(row, "id"),
revision: readString(row, "revision"),
service: readString(row, "service"),
connectionName: readString(row, "connection_name"),
credential: parseJson<ResolvedCredential>(await this.secretCodec.decode(readString(row, "value"))),
})),
);
}
}
class PostgresOAuthClientConfigStore implements IOAuthClientConfigStore {
private readonly pool: Pool;
private readonly secretCodec: ISecretCodec;
constructor(pool: Pool, secretCodec: ISecretCodec) {
this.pool = pool;
this.secretCodec = secretCodec;
}
async get(service: string): Promise<OAuthClientConfig | undefined> {
const result = await this.pool.query<RuntimeRow>("select value from oauth_client_configs where service = $1", [
service,
]);
const row = result.rows[0];
return row ? parseJson<OAuthClientConfig>(await this.secretCodec.decode(readString(row, "value"))) : undefined;
}
async set(config: OAuthClientConfig): Promise<void> {
await this.pool.query(
`
insert into oauth_client_configs (service, value, updated_at)
values ($1, $2, $3)
on conflict(service) do update set value = excluded.value, updated_at = excluded.updated_at
`,
[config.service, await this.secretCodec.encode(JSON.stringify(config)), new Date().toISOString()],
);
}
async delete(service: string): Promise<void> {
await this.pool.query("delete from oauth_client_configs where service = $1", [service]);
}
async list(): Promise<OAuthClientConfig[]> {
const result = await this.pool.query<RuntimeRow>("select value from oauth_client_configs order by service");
return await Promise.all(
result.rows.map(async (row) =>
parseJson<OAuthClientConfig>(await this.secretCodec.decode(readString(row, "value"))),
),
);
}
}
class PostgresOAuthStateStore implements IOAuthStateStore {
private readonly pool: Pool;
private readonly secretCodec: ISecretCodec;
constructor(pool: Pool, secretCodec: ISecretCodec) {
this.pool = pool;
this.secretCodec = secretCodec;
}
async set(state: OAuthAuthorizationState): Promise<void> {
await this.pool.query(
`
insert into oauth_states (state, value, created_at)
values ($1, $2, $3)
on conflict(state) do update set value = excluded.value, created_at = excluded.created_at
`,
[state.state, await this.secretCodec.encode(JSON.stringify(state)), state.createdAt],
);
}
async take(state: string): Promise<OAuthAuthorizationState | undefined> {
const result = await this.pool.query<RuntimeRow>("delete from oauth_states where state = $1 returning value", [
state,
]);
const row = result.rows[0];
return row
? parseJson<OAuthAuthorizationState>(await this.secretCodec.decode(readString(row, "value")))
: undefined;
}
}
class PostgresRuntimeTokenStore implements IRuntimeTokenStore {
private readonly pool: Pool;
constructor(pool: Pool) {
this.pool = pool;
}
async add(record: RuntimeTokenRecord): Promise<void> {
await this.pool.query(
`
insert into runtime_tokens (
id, name, token_hash, allowed_actions, blocked_actions, allowed_proxies, allowed_connections, created_at, last_used_at
)
values ($1, $2, $3, $4, $5, $6, $7, $8, $9)
`,
[
record.id,
record.name,
record.tokenHash,
JSON.stringify(record.allowedActions),
JSON.stringify(record.blockedActions),
JSON.stringify(record.allowedProxies),
JSON.stringify(record.allowedConnections ?? []),
record.createdAt,
record.lastUsedAt ?? null,
],
);
}
async list(): Promise<RuntimeTokenRecord[]> {
const result = await this.pool.query<RuntimeRow>(`
select id, name, token_hash, allowed_actions, blocked_actions, allowed_proxies, allowed_connections, created_at, last_used_at
from runtime_tokens
where revoked_at is null
order by created_at desc, id desc
`);
return result.rows.map(readRuntimeTokenRow);
}
async findByHash(tokenHash: string): Promise<RuntimeTokenRecord | undefined> {
const result = await this.pool.query<RuntimeRow>(
`
select id, name, token_hash, allowed_actions, blocked_actions, allowed_proxies, allowed_connections, created_at, last_used_at
from runtime_tokens
where token_hash = $1 and revoked_at is null
`,
[tokenHash],
);
const row = result.rows[0];
return row ? readRuntimeTokenRow(row) : undefined;
}
async updatePolicy(id: string, policy: TokenPolicy): Promise<RuntimeTokenRecord | undefined> {
const result = await this.pool.query<RuntimeRow>(
`
update runtime_tokens
set allowed_actions = $1, blocked_actions = $2, allowed_proxies = $3, allowed_connections = $4
where id = $5 and revoked_at is null
returning id, name, token_hash, allowed_actions, blocked_actions, allowed_proxies, allowed_connections, created_at, last_used_at
`,
[
JSON.stringify(policy.allowedActions),
JSON.stringify(policy.blockedActions),
JSON.stringify(policy.allowedProxies),
JSON.stringify(policy.allowedConnections ?? []),
id,
],
);
const row = result.rows[0];
return row ? readRuntimeTokenRow(row) : undefined;
}
async revoke(id: string): Promise<boolean> {
const result = await this.pool.query("delete from runtime_tokens where id = $1", [id]);
return (result.rowCount ?? 0) > 0;
}
async markUsed(id: string, usedAt: string): Promise<void> {
await this.pool.query("update runtime_tokens set last_used_at = $1 where id = $2 and revoked_at is null", [
usedAt,
id,
]);
}
}
class PostgresRuntimePolicyStore implements IRuntimePolicyStore {
private readonly pool: Pool;
constructor(pool: Pool) {
this.pool = pool;
}
async get(): Promise<RuntimePolicyRecord | undefined> {
const result = await this.pool.query<RuntimeRow>("select value, updated_at from runtime_policy where id = 1");
const row = result.rows[0];
return row
? {
rules: parseJson(readString(row, "value")),
updatedAt: readString(row, "updated_at"),
}
: undefined;
}
async set(record: RuntimePolicyRecord): Promise<void> {
await this.pool.query(
`
insert into runtime_policy (id, value, updated_at)
values (1, $1, $2)
on conflict(id) do update set value = excluded.value, updated_at = excluded.updated_at
`,
[JSON.stringify(record.rules), record.updatedAt],
);
}
}
class PostgresIdempotencyStore implements IIdempotencyStore {
private readonly pool: Pool;
private readonly secretCodec: ISecretCodec;
constructor(pool: Pool, secretCodec: ISecretCodec) {
this.pool = pool;
this.secretCodec = secretCodec;
}
async claim(input: IdempotencyClaimInput): Promise<IdempotencyClaimResult> {
const claim = await runInTransaction(this.pool, async (client) => {
await client.query("delete from idempotency_records where expires_at <= $1", [input.now]);
const inserted = await client.query(
`
insert into idempotency_records (
key_hash, claim_id, request_hash, state, response_value, created_at, expires_at
)
values ($1, $2, $3, 'in_progress', null, $4, $5)
on conflict(key_hash) do nothing
returning key_hash
`,
[input.keyHash, input.claimId, input.requestHash, input.now, input.expiresAt],
);
if ((inserted.rowCount ?? 0) > 0) {
return { kind: "acquired" } as const;
}
const existing = await client.query<RuntimeRow>(
"select request_hash, state, response_value from idempotency_records where key_hash = $1",
[input.keyHash],
);
return { kind: "existing", row: existing.rows[0]! } as const;
});
if (claim.kind === "acquired") {
return claim;
}
if (readString(claim.row, "request_hash") !== input.requestHash) {
return { kind: "conflict" };
}
if (readString(claim.row, "state") === "in_progress") {
return { kind: "in_progress" };
}
return {
kind: "completed",
response: parseRuntimeActionHttpResult(
parseJson(await this.secretCodec.decode(readString(claim.row, "response_value"))),
),
};
}
async complete(input: CompleteIdempotencyInput): Promise<boolean> {
const result = await this.pool.query(
`
update idempotency_records
set state = 'completed', response_value = $1, expires_at = $2
where key_hash = $3
and claim_id = $4
and request_hash = $5
and state = 'in_progress'
`,
[
await this.secretCodec.encode(JSON.stringify(input.response)),
input.expiresAt,
input.keyHash,
input.claimId,
input.requestHash,
],
);
return (result.rowCount ?? 0) > 0;
}
}
class PostgresRunLogStore implements IRunLogStore {
private readonly pool: Pool;
private readonly limit: number;
constructor(pool: Pool, limit: number) {
this.pool = pool;
this.limit = limit;
}
async add(run: RunLog): Promise<RunLogWriteResult> {
await this.pool.query(
`
insert into runs (id, service, action_id, caller, started_at, completed_at, ok, value)
values ($1, $2, $3, $4, $5, $6, $7, $8)
on conflict(id) do update set
service = excluded.service,
action_id = excluded.action_id,
caller = excluded.caller,
started_at = excluded.started_at,
completed_at = excluded.completed_at,
ok = excluded.ok,
value = excluded.value
`,
[
run.id,
run.service,
run.actionId,
run.caller,
run.startedAt,
run.completedAt,
run.ok ? 1 : 0,
JSON.stringify(run),
],
);
try {
await this.pool.query(
`
delete from runs
where id in (
select id from runs
order by started_at desc, id desc
offset $1
)
`,
[this.limit],
);
return { retentionApplied: true };
} catch {
return { retentionApplied: false };
}
}
async get(id: string): Promise<RunLog | undefined> {
const result = await this.pool.query<RuntimeRow>("select service, value from runs where id = $1", [id]);
const row = result.rows[0];
return row ? readRunLogRow(row) : undefined;
}
async list(input: RunLogListInput = {}): Promise<RunLogPage> {
const limit = Math.max(1, Math.min(input.limit ?? this.limit, this.limit));
const cursor = decodeRunLogCursor(input.cursor);
const conditions: string[] = [];
const values: Array<string | number> = [];
if (cursor) {
const startedAtParameter = values.push(cursor.startedAt);
const repeatedStartedAtParameter = values.push(cursor.startedAt);
const idParameter = values.push(cursor.id);
conditions.push(
`(started_at < $${startedAtParameter} or (started_at = $${repeatedStartedAtParameter} and id < $${idParameter}))`,
);
}
if (input.service) {
conditions.push(`service = $${values.push(input.service)}`);
}
if (input.actionId) {
conditions.push(`action_id = $${values.push(input.actionId)}`);
}
if (input.caller) {
conditions.push(`caller = $${values.push(input.caller)}`);
}
if (input.ok !== undefined) {
conditions.push(`ok = $${values.push(input.ok ? 1 : 0)}`);
}
const where = conditions.length > 0 ? `where ${conditions.join(" and ")}` : "";
const limitParameter = values.push(limit + 1);
const result = await this.pool.query<RuntimeRow>(
`select service, value from runs ${where} order by started_at desc, id desc limit $${limitParameter}`,
values,
);
const runs = result.rows.map(readRunLogRow);
const items = runs.slice(0, limit);
return {
items,
nextCursor: runs.length > limit && items.length > 0 ? encodeRunLogCursor(items[items.length - 1]) : undefined,
};
}
}
function readRuntimeTokenRow(row: RuntimeRow): RuntimeTokenRecord {
return {
id: readString(row, "id"),
name: readString(row, "name"),
tokenHash: readString(row, "token_hash"),
allowedActions: parseJson(readString(row, "allowed_actions")),
blockedActions: parseJson(readString(row, "blocked_actions")),
allowedProxies: parseJson(readString(row, "allowed_proxies")),
allowedConnections: parseJson(readOptionalString(row, "allowed_connections") ?? "[]"),
createdAt: readString(row, "created_at"),
lastUsedAt: readOptionalString(row, "last_used_at"),
};
}
function readRunLogRow(row: RuntimeRow): RunLog {
const run = parseJson<RunLog>(readString(row, "value"));
return { ...run, service: readString(row, "service") };
}
function readString(row: RuntimeRow, key: string): string {
const value = row[key];
if (typeof value !== "string") {
throw new Error(`Expected PostgreSQL column ${key} to be a string.`);
}
return value;
}
function readOptionalString(row: RuntimeRow, key: string): string | undefined {
const value = row[key];
if (value == null) {
return undefined;
}
if (typeof value !== "string") {
throw new Error(`Expected PostgreSQL column ${key} to be a string.`);
}
return value;
}
function parseJson<T>(value: string): T {
return JSON.parse(value) as T;
}
async function runInTransaction<T>(pool: Pool, work: (client: PoolClient) => Promise<T>): Promise<T> {
const client = await pool.connect();
try {
await client.query("begin");
const result = await work(client);
await client.query("commit");
return result;
} catch (error) {
await client.query("rollback");
throw error;
} finally {
client.release();
}
}