From 9cb376b032bd21a2441ba510f8b5919db949c86e Mon Sep 17 00:00:00 2001 From: mia <35839923+distanceqo@users.noreply.github.com> Date: Mon, 10 Aug 2026 22:08:33 +0800 Subject: [PATCH 1/5] feat: add Langfuse Wasm FDW MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reads trace and observation data from Langfuse, an open source LLM observability platform, so LLM spend can be attributed to rows in a local users table in SQL. Targets the Langfuse Public API rather than the ClickHouse store behind it, which upstream documents as an unstable contract and which Langfuse Cloud users cannot reach at all. Supports select on `traces` and `observations`, import foreign schema, limit and equality pushdown, and time bounds via fromTimestamp/toTimestamp (or fromStartTime/toStartTime on observations — the column and params differ per endpoint). Credentials resolve from a Vault secret name, a secret id, or a plaintext option. Usage and cost arrive as objects keyed by metric name and are flattened into scalar columns so aggregates need no JSON extraction. The fallback chains are load-bearing: `observations` returns calculatedTotalCost with no totalCost key, and promptTokens rather than usageDetails.input. Verified end to end on Supabase against a live Langfuse Cloud project. --- docs/catalog/index.md | 2 + docs/catalog/langfuse.md | 367 +++++++++++++ docs/catalog/wasm/index.md | 12 + mkdocs.yaml | 1 + wasm-wrappers/fdw/Cargo.lock | 185 ++++--- wasm-wrappers/fdw/Cargo.toml | 1 + wasm-wrappers/fdw/langfuse_fdw/Cargo.toml | 25 + wasm-wrappers/fdw/langfuse_fdw/src/lib.rs | 534 +++++++++++++++++++ wasm-wrappers/fdw/langfuse_fdw/wit/world.wit | 10 + 9 files changed, 1076 insertions(+), 61 deletions(-) create mode 100644 docs/catalog/langfuse.md create mode 100644 wasm-wrappers/fdw/langfuse_fdw/Cargo.toml create mode 100644 wasm-wrappers/fdw/langfuse_fdw/src/lib.rs create mode 100644 wasm-wrappers/fdw/langfuse_fdw/wit/world.wit diff --git a/docs/catalog/index.md b/docs/catalog/index.md index c23864b76..e55745005 100644 --- a/docs/catalog/index.md +++ b/docs/catalog/index.md @@ -27,6 +27,7 @@ Each FDW documentation includes a detailed "Limitations" section that describes | HubSpot | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ | | Infura | ✅ | ❌ | ❌ | ❌ | ❌ | ✅ | | Iceberg | ✅ | ✅ | ❌ | ❌ | ❌ | ❌ | +| Langfuse | ✅ | ❌ | ❌ | ❌ | ❌ | ✅ | | Logflare | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ | | MongoDB | ✅ | ✅ | ✅ | ✅ | ❌ | ✅ | | MySQL | ✅ | ✅ | ✅ | ✅ | ❌ | ✅ | @@ -57,6 +58,7 @@ See [Developing a Wasm Wrapper](../guides/create-wasm-wrapper.md) for instructio | Gravatar | [Automattic](https://automattic.com) | [Link](gravatar.md) | [Link](https://github.com/Automattic/gravatar-wasm-fdw) | | HubSpot | [Supabase](https://supabase.com) | [Link](hubspot.md) | [Link](https://github.com/supabase/wrappers/tree/main/wasm-wrappers/fdw/hubspot_fdw) | | Infura | [Supabase](https://supabase.com) | [Link](infura.md) | [Link](https://github.com/supabase/wrappers/tree/main/wasm-wrappers/fdw/infura_fdw) | +| Langfuse | [Supabase](https://supabase.com) | [Link](langfuse.md) | [Link](https://github.com/supabase/wrappers/tree/main/wasm-wrappers/fdw/langfuse_fdw) | | Notion | [Supabase](https://supabase.com) | [Link](notion.md) | [Link](https://github.com/supabase/wrappers/tree/main/wasm-wrappers/fdw/notion_fdw) | | OpenAPI | [Cody Bromley](https://github.com/codybrom) | [Link](openapi.md) | [Link](https://github.com/supabase/wrappers/tree/main/wasm-wrappers/fdw/openapi_fdw) | | Orb | [Supabase](https://supabase.com) | [Link](orb.md) | [Link](https://github.com/supabase/wrappers/tree/main/wasm-wrappers/fdw/orb_fdw) | diff --git a/docs/catalog/langfuse.md b/docs/catalog/langfuse.md new file mode 100644 index 000000000..fde6f4c54 --- /dev/null +++ b/docs/catalog/langfuse.md @@ -0,0 +1,367 @@ +--- +source: +documentation: +author: supabase +tags: + - wasm + - official +--- + +# Langfuse + +[Langfuse](https://langfuse.com/) is an open source LLM observability platform which records traces, token usage, and cost for LLM applications. + +The Langfuse Wrapper is a WebAssembly(Wasm) foreign data wrapper which allows you to read trace and observation data from Langfuse for use within your Postgres database. + +It targets the Langfuse [Public API](https://langfuse.com/docs/api) rather than the ClickHouse store behind it, which upstream documents as [not a stable API contract](https://langfuse.com/self-hosting/infrastructure/clickhouse) and which Langfuse Cloud users cannot reach at all. + +## Available Versions + +| Version | Wasm Package URL | Checksum | Required Wrappers Version | +| ------- | --------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------ | ------------------------- | +| 0.1.0 | `https://github.com/supabase/wrappers/releases/download/wasm_langfuse_fdw_v0.1.0/langfuse_fdw.wasm` | `` | >=0.5.0 | + +## Preparation + +Before you can query Langfuse, you need to enable the Wrappers extension and store your credentials in Postgres. + +### Enable Wrappers + +Make sure the `wrappers` extension is installed on your database: + +```sql +create extension if not exists wrappers with schema extensions; +``` + +### Enable the Langfuse Wrapper + +Enable the Wasm foreign data wrapper: + +```sql +create foreign data wrapper wasm_wrapper + handler wasm_fdw_handler + validator wasm_fdw_validator; +``` + +### Store your credentials (optional) + +By default, Postgres stores FDW credentials inside `pg_catalog.pg_foreign_server` in plain text. Anyone with access to this table will be able to view these credentials. Wrappers is designed to work with [Vault](https://supabase.com/docs/guides/database/vault), which provides an additional level of security for storing credentials. We recommend using Vault to store your credentials. + +Langfuse authenticates with a public/secret key pair, both found under project settings. + +```sql +-- Save your Langfuse keys in Vault +select vault.create_secret( + '', -- pk-lf-... + 'langfuse_public_key', + 'Langfuse public key for Wrappers' +); + +select vault.create_secret( + '', -- sk-lf-... + 'langfuse_secret_key', + 'Langfuse secret key for Wrappers' +); +``` + +### Connecting to Langfuse + +We need to provide Postgres with the credentials to access Langfuse and any additional options. We can do this using the `create server` command: + +=== "With Vault" + + ```sql + create server langfuse_server + foreign data wrapper wasm_wrapper + options ( + fdw_package_url 'https://github.com/supabase/wrappers/releases/download/wasm_langfuse_fdw_v0.1.0/langfuse_fdw.wasm', + fdw_package_name 'supabase:langfuse-fdw', + fdw_package_version '0.1.0', + fdw_package_checksum '', + api_url 'https://cloud.langfuse.com', -- optional + public_key_name 'langfuse_public_key', -- the Vault secret name from above + secret_key_name 'langfuse_secret_key' + ); + ``` + +=== "Without Vault" + + ```sql + create server langfuse_server + foreign data wrapper wasm_wrapper + options ( + fdw_package_url 'https://github.com/supabase/wrappers/releases/download/wasm_langfuse_fdw_v0.1.0/langfuse_fdw.wasm', + fdw_package_name 'supabase:langfuse-fdw', + fdw_package_version '0.1.0', + fdw_package_checksum '', + api_url 'https://cloud.langfuse.com', -- optional + public_key '', + secret_key '' + ); + ``` + +Note the `fdw_package_*` options are required, which specify the Wasm package metadata. You can get the available package version list from [above](#available-versions). + +!!! warning + + API keys are bound to the region the Langfuse project was created in. `api_url` must match it or every request returns 401. Use `https://cloud.langfuse.com` for EU, `https://us.cloud.langfuse.com` for US, `https://jp.cloud.langfuse.com` for Japan, or your own URL when self-hosting. + +### Create a schema + +We recommend creating a schema to hold all the foreign tables: + +```sql +create schema if not exists langfuse; +``` + +## Options + +The full list of foreign table options are below: + +- `object` - API path after `/api/public/`, required. For example `traces` or `observations`. +- `fields` - Field groups to request, optional. Applies to the `v2/` endpoints only. + +The full list of server options are below: + +- `api_url` - Langfuse API base URL, optional. Defaults to `https://cloud.langfuse.com`. +- `public_key_name` / `secret_key_name` - Vault secret names holding the keys. +- `public_key_id` / `secret_key_id` - Vault secret ids, as an alternative to the names. +- `public_key` / `secret_key` - Plaintext keys, for local development. +- `page_size` - Rows to request per upstream call, optional. Defaults to `100`, maximum `1000`. +- `verbose` - Set to `'true'` to log each request URL as an `INFO` message, optional. + +## Entities + +We can use SQL [import foreign schema](https://www.postgresql.org/docs/current/sql-importforeignschema.html) to import foreign table definitions from Langfuse. + +For example, using below SQL can automatically create foreign tables in the `langfuse` schema. + +```sql +-- create all the foreign tables +import foreign schema langfuse from server langfuse_server into langfuse; +``` + +### Traces + +A trace is one end-to-end request through the application. Traces carry `user_id` and an aggregate cost, which makes this the table to join local user tables against. + +Ref: [Langfuse data model](https://langfuse.com/docs/observability/data-model) + +#### Operations + +| Object | Select | Insert | Update | Delete | Truncate | +| --------------------- | :----: | :----: | :----: | :----: | :------: | +| traces | ✅ | ❌ | ❌ | ❌ | ❌ | + +#### Usage + +```sql +create foreign table langfuse.traces ( + id text, + name text, + user_id text, + session_id text, + environment text, + release text, + version text, + total_cost double precision, + latency double precision, + timestamp timestamp, + created_at timestamp, + updated_at timestamp, + input text, + output text, + metadata jsonb, + tags jsonb +) + server langfuse_server + options ( + object 'traces', + rowid_column 'id' + ); +``` + +!!! note + + You can use `import foreign schema` statement to automatically create the foreign tables [see above](#entities) + +### Observations + +An observation is a single step inside a trace, most usefully a model call, with token counts and per-call cost. + +#### Operations + +| Object | Select | Insert | Update | Delete | Truncate | +| --------------------- | :----: | :----: | :----: | :----: | :------: | +| observations | ✅ | ❌ | ❌ | ❌ | ❌ | + +#### Usage + +```sql +create foreign table langfuse.observations ( + id text, + trace_id text, + type text, + name text, + level text, + model text, + input_tokens bigint, + output_tokens bigint, + total_tokens bigint, + input_cost double precision, + output_cost double precision, + total_cost double precision, + latency double precision, + time_to_first_token double precision, + start_time timestamp, + end_time timestamp, + completion_start_time timestamp, + prompt_name text, + prompt_version bigint, + input text, + output text, + metadata jsonb, + model_parameters jsonb +) + server langfuse_server + options ( + object 'observations', + rowid_column 'id' + ); +``` + +#### Notes + +- Column names are snake_case and mapped to the API's camelCase automatically, so `session_id` reads `sessionId`. A column the API does not return is NULL rather than an error, so you can declare only the columns you need. + +- Langfuse returns usage and cost as objects keyed by metric name. These are flattened into scalar columns so aggregates work without JSON extraction: + + | Column | Source | + | ---------------- | --------------------------------------------------------- | + | `total_tokens` | `usageDetails.total` | + | `input_tokens` | `usageDetails.input`, else `promptTokens` | + | `output_tokens` | `usageDetails.output`, else `completionTokens` | + | `total_cost` | `costDetails.total`, else `calculatedTotalCost` | + | `input_cost` | `costDetails.input`, else `calculatedInputCost` | + | `output_cost` | `costDetails.output`, else `calculatedOutputCost` | + + The fallbacks matter in practice: the `observations` endpoint returns cost as `calculatedTotalCost` and has no `totalCost` key at all. + +- `observations` does not return `user_id`, which lives on the parent trace, although it is still accepted as a filter. Join through `trace_id` to attribute a call to a user. + +- `input`, `output`, and `metadata` hold arbitrary JSON. Declare them as `jsonb` to query into them, or as `text` for the raw value. + +## Query Pushdown Support + +This FDW supports: + +- `limit` pushdown, so `limit 10` costs a single upstream request regardless of project size +- `where` equality pushdown on `trace_id`, `user_id`, `session_id`, `type`, `level`, and `name` +- `where` time bounds using `>=` and `<` + +Pages are fetched lazily as the scan consumes them, rather than up front. + +The time column and its parameters differ per endpoint: `observations` filters `start_time` via `fromStartTime`/`toStartTime`, everything else filters `timestamp` via `fromTimestamp`/`toTimestamp`. + +Time bounds need literal timestamps. Postgres only forwards quals it can evaluate up front, so `now()` arrives as an empty qual list and the filter runs locally after every page has been fetched: + +```sql +-- pushed down +select * from langfuse.traces +where timestamp >= '2026-08-02'::timestamp + and timestamp < '2026-08-09'::timestamp; + +-- not pushed down; fetches every page, then filters +select * from langfuse.traces +where timestamp >= now() - interval '7 days'; +``` + +Confirm with `explain (verbose)` and read the `Wrappers: quals` line — an empty list means nothing was pushed down. + +## Supported Data Types + +| Postgres Data Type | Langfuse Data Type | +| ------------------ | ------------------ | +| boolean | Boolean | +| bigint | Number | +| double precision | Number | +| text | String | +| timestamp | Time | +| jsonb | Json | + +The Langfuse API uses JSON formatted data, please refer to [Langfuse API docs](https://langfuse.com/docs/api) for more details. + +## Limitations + +This section describes important limitations and considerations when using this FDW: + +- Read only; no `insert`, `update`, or `delete` support +- Only `>=` and `<` time bounds are pushed down. The API's bounds are inclusive-from and exclusive-to, so `>` and `<=` would need an epsilon shift to stay correct and are left to Postgres +- `now()` and other non-constant expressions in a time filter are not pushed down; use literal timestamps +- The `v2/observations` endpoint reads a different store than `observations`. On a freshly created cloud project, rows ingested through both `/api/public/ingestion` and the OTLP endpoint were readable via `observations` while `v2/observations` returned an empty result, although it still validated query parameters. Prefer `observations` until that settles +- `re_scan` is not supported, so these tables cannot sit on the inner side of a nested-loop join. Materialize with a CTE if needed +- Materialized views using these foreign tables may fail during logical backups + +## Examples + +Below are some examples on how to use Langfuse foreign tables. + +### Basic example + +```sql +import foreign schema langfuse from server langfuse_server into langfuse; + +select id, name, user_id, total_cost +from langfuse.traces +limit 10; +``` + +### Cost per user + +```sql +select user_id, + count(*) as traces, + sum(total_cost) as cost_usd, + round(avg(latency)::numeric, 2) as avg_latency_s +from langfuse.traces +group by user_id +order by cost_usd desc; +``` + +### Token usage by model + +```sql +select model, + count(*) as calls, + sum(input_tokens) as input_tokens, + sum(output_tokens) as output_tokens, + sum(total_cost) as cost_usd +from langfuse.observations +group by model +order by cost_usd desc; +``` + +### Joining against a local users table + +This is the query the wrapper exists for: attributing LLM spend to rows in your own database. + +Aggregate the foreign table separately before joining. A direct join against several local tables multiplies the trace rows and inflates `sum()`. + +```sql +with llm as ( + select user_id, + count(*) as llm_calls, + sum(total_cost) as cost_usd + from langfuse.traces + where timestamp >= '2026-08-01'::timestamp + group by user_id +) +select u.email, + coalesce(l.llm_calls, 0) as llm_calls, + l.cost_usd +from auth.users u +left join llm l on l.user_id = u.id::text +order by l.cost_usd desc nulls last; +``` + +This assumes the application passes the Supabase user id to Langfuse as its `userId`. If the two systems use different identifiers, join through a mapping table instead. diff --git a/docs/catalog/wasm/index.md b/docs/catalog/wasm/index.md index 6ac99d6f3..792302a2c 100644 --- a/docs/catalog/wasm/index.md +++ b/docs/catalog/wasm/index.md @@ -97,6 +97,18 @@ Foreign data wrappers built with Wasm which can be used on Supabase platform. :octicons-code-24: [source](https://github.com/supabase/wrappers/tree/wasm_infura_fdw_v0.2.0/wasm-wrappers/fdw/infura_fdw)   :material-file-document: [docs](../infura.md) +- :simple-webassembly:   **[Langfuse](../langfuse.md)** + + ---- + + Foreign data wrapper for [Langfuse](https://langfuse.com/) LLM observability data. + + Supported by [Supabase](https://www.supabase.com) + + :octicons-tag-24: [v0.1.0](https://github.com/supabase/wrappers/releases/tag/wasm_langfuse_fdw_v0.1.0)   + :octicons-code-24: [source](https://github.com/supabase/wrappers/tree/wasm_langfuse_fdw_v0.1.0/wasm-wrappers/fdw/langfuse_fdw)   + :material-file-document: [docs](../langfuse.md) + - :simple-webassembly:   **[Notion](../notion.md)** ---- diff --git a/mkdocs.yaml b/mkdocs.yaml index 2a20f42a5..8b7bdcb55 100644 --- a/mkdocs.yaml +++ b/mkdocs.yaml @@ -41,6 +41,7 @@ nav: - Gravatar: "catalog/gravatar.md" - HubSpot: "catalog/hubspot.md" - Infura: "catalog/infura.md" + - Langfuse: "catalog/langfuse.md" - Notion: "catalog/notion.md" - OpenAPI: "catalog/openapi.md" - Orb: "catalog/orb.md" diff --git a/wasm-wrappers/fdw/Cargo.lock b/wasm-wrappers/fdw/Cargo.lock index d4138e0ed..ec86ea46b 100644 --- a/wasm-wrappers/fdw/Cargo.lock +++ b/wasm-wrappers/fdw/Cargo.lock @@ -4,24 +4,30 @@ version = 4 [[package]] name = "android_system_properties" -version = "0.1.5" +version = "0.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311" +checksum = "ae221649c9976a6f6c56ae1facf410f3ddb33cc661c4b7b61020a912d4237fbc" dependencies = [ "libc", ] [[package]] name = "autocfg" -version = "1.5.0" +version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" +checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" + +[[package]] +name = "base64" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" [[package]] name = "bumpalo" -version = "3.19.1" +version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5dd9dc738b7a8311c7ade152424974d8115f2cdad61e8dab8dac9f2362298510" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" [[package]] name = "cal_fdw" @@ -43,9 +49,9 @@ dependencies = [ [[package]] name = "cc" -version = "1.2.56" +version = "1.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "aebf35691d1bfb0ac386a69bac2fde4dd276fb618cf8bf4f5318fe285e821bb2" +checksum = "5d262e149917187838d5b42777c8253bcb64500067342904e7d429499a6f277e" dependencies = [ "find-msvc-tools", "shlex", @@ -67,9 +73,9 @@ checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" [[package]] name = "chrono" -version = "0.4.43" +version = "0.4.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fac4744fb15ae8337dc853fee7fb3f4e48c0fbaa23d0afe49c447b4fab126118" +checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327" dependencies = [ "iana-time-zone", "js-sys", @@ -100,15 +106,39 @@ checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" [[package]] name = "find-msvc-tools" -version = "0.1.9" +version = "0.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "26b73573e6edcd2af0cdf47bd6cb58f0b3839491263c314eaad1ccf24430e1de" + +[[package]] +name = "futures-core" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2cd50c473c80f6d7c3670a752354b8e569b1a7cbfdc0419ec88e5edad85e0dc7" + +[[package]] +name = "futures-task" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b231ed28831efb4a61a08580c4bc233ec56bc009f4cd8f52da2c3cb97df0c109" + +[[package]] +name = "futures-util" +version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +checksum = "a77a90a256fce34da66415271e30f94ee91c57b04b8a2c042d9cf3220179deaa" +dependencies = [ + "futures-core", + "futures-task", + "pin-project-lite", + "slab", +] [[package]] name = "hashbrown" -version = "0.16.1" +version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" [[package]] name = "helloworld_fdw" @@ -151,9 +181,9 @@ dependencies = [ [[package]] name = "indexmap" -version = "2.13.0" +version = "2.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7714e70437a7dc3ac8eb7e6f8df75fd8eb422675fc7678aff7364301092b1017" +checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" dependencies = [ "equivalent", "hashbrown", @@ -169,37 +199,47 @@ dependencies = [ [[package]] name = "itoa" -version = "1.0.17" +version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "92ecc6618181def0457392ccd0ee51198e065e016d1d527a7ac1b6dc7c1f09d2" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" [[package]] name = "js-sys" -version = "0.3.85" +version = "0.3.104" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8c942ebf8e95485ca0d52d97da7c5a2c387d0e7f0ba4c35e93bfcaee045955b3" +checksum = "0e0c1080212aad755ea003d18543e8768dd432c48819efd73a7bf1e39b7a5a3a" dependencies = [ - "once_cell", + "cfg-if", + "futures-util", "wasm-bindgen", ] +[[package]] +name = "langfuse_fdw" +version = "0.1.0" +dependencies = [ + "base64", + "serde_json", + "wit-bindgen-rt", +] + [[package]] name = "libc" -version = "0.2.182" +version = "0.2.189" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6800badb6cb2082ffd7b6a67e6125bb39f18782f793520caee8cb8846be06112" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" [[package]] name = "log" -version = "0.4.29" +version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" [[package]] name = "memchr" -version = "2.8.0" +version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" [[package]] name = "notion_fdw" @@ -221,9 +261,9 @@ dependencies = [ [[package]] name = "once_cell" -version = "1.21.3" +version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" [[package]] name = "openapi_fdw" @@ -253,29 +293,35 @@ dependencies = [ "wit-bindgen-rt", ] +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + [[package]] name = "proc-macro2" -version = "1.0.106" +version = "1.0.107" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" dependencies = [ "unicode-ident", ] [[package]] name = "quote" -version = "1.0.44" +version = "1.0.47" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "21b2ebcf727b7760c461f091f9f0f539b77b8e87f2fd88131e7f1b433b3cece4" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" dependencies = [ "proc-macro2", ] [[package]] name = "rustversion" -version = "1.0.22" +version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" +checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" [[package]] name = "ryu" @@ -285,9 +331,9 @@ checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" [[package]] name = "serde" -version = "1.0.228" +version = "1.0.229" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" dependencies = [ "serde_core", "serde_derive", @@ -295,29 +341,29 @@ dependencies = [ [[package]] name = "serde_core" -version = "1.0.228" +version = "1.0.229" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" dependencies = [ "serde_derive", ] [[package]] name = "serde_derive" -version = "1.0.228" +version = "1.0.229" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 3.0.3", ] [[package]] name = "serde_json" -version = "1.0.149" +version = "1.0.151" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "83fc039473c5595ace860d8c4fafa220ff474b3fc6bfdb4293327f1a37e94d86" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" dependencies = [ "itoa", "memchr", @@ -341,9 +387,9 @@ dependencies = [ [[package]] name = "shlex" -version = "1.3.0" +version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" [[package]] name = "shopify_fdw" @@ -354,6 +400,12 @@ dependencies = [ "wit-bindgen-rt", ] +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + [[package]] name = "slack_fdw" version = "0.2.0" @@ -373,9 +425,20 @@ dependencies = [ [[package]] name = "syn" -version = "2.0.116" +version = "2.0.119" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3df424c70518695237746f84cede799c9c58fcb37450d7b23716568cc8bc69cb" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" dependencies = [ "proc-macro2", "quote", @@ -402,9 +465,9 @@ checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da" [[package]] name = "wasm-bindgen" -version = "0.2.108" +version = "0.2.127" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "64024a30ec1e37399cf85a7ffefebdb72205ca1c972291c51512360d90bd8566" +checksum = "1b70935747edd64d89de3efa29d73789b806c15798f8e7dca4d8ac356b50ce70" dependencies = [ "cfg-if", "once_cell", @@ -415,9 +478,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.108" +version = "0.2.127" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "008b239d9c740232e71bd39e8ef6429d27097518b6b30bdf9086833bd5b6d608" +checksum = "77775f8f3f7217702089053b94958f8f54061a3f663417df76e19cbdcca29bc1" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -425,22 +488,22 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.108" +version = "0.2.127" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5256bae2d58f54820e6490f9839c49780dff84c65aeab9e772f15d5f0e913a55" +checksum = "e11d33f857dc2fb11b8bc75aee111aa9cbeb12cd9f25efd3d4c2a3dd4e235284" dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn", + "syn 2.0.119", "wasm-bindgen-shared", ] [[package]] name = "wasm-bindgen-shared" -version = "0.2.108" +version = "0.2.127" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1f01b580c9ac74c8d8f0c0e4afb04eeef2acf145458e52c03845ee9cd23e3d12" +checksum = "7ef64dbcc55df09c7e5a46182d181c2cfa3e925f3da937ea764728b4bbb9dcbf" dependencies = [ "unicode-ident", ] @@ -466,7 +529,7 @@ checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.119", ] [[package]] @@ -477,7 +540,7 @@ checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.119", ] [[package]] @@ -512,6 +575,6 @@ checksum = "c4db52a11d4dfb0a59f194c064055794ee6564eb1ced88c25da2cf76e50c5621" [[package]] name = "zmij" -version = "1.0.21" +version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/wasm-wrappers/fdw/Cargo.toml b/wasm-wrappers/fdw/Cargo.toml index e0a9b9316..51d83a084 100644 --- a/wasm-wrappers/fdw/Cargo.toml +++ b/wasm-wrappers/fdw/Cargo.toml @@ -7,6 +7,7 @@ members = [ "helloworld_fdw", "hubspot_fdw", "infura_fdw", + "langfuse_fdw", "notion_fdw", "openapi_fdw", "orb_fdw", diff --git a/wasm-wrappers/fdw/langfuse_fdw/Cargo.toml b/wasm-wrappers/fdw/langfuse_fdw/Cargo.toml new file mode 100644 index 000000000..465fc5ae5 --- /dev/null +++ b/wasm-wrappers/fdw/langfuse_fdw/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "langfuse_fdw" +version = "0.1.0" +edition = { workspace = true } +homepage = { workspace = true } +rust-version = { workspace = true } + +[lib] +crate-type = ["cdylib"] + +[dependencies] +wit-bindgen-rt = "0.41.0" +serde_json = "1.0" +base64 = "0.22" + +[package.metadata.component] +package = "supabase:langfuse-fdw" + +[package.metadata.component.dependencies] + +[package.metadata.component.target] +path = "wit" + +[package.metadata.component.target.dependencies] +"supabase:wrappers" = { path = "../../wit/v2" } diff --git a/wasm-wrappers/fdw/langfuse_fdw/src/lib.rs b/wasm-wrappers/fdw/langfuse_fdw/src/lib.rs new file mode 100644 index 000000000..ac5c34ef6 --- /dev/null +++ b/wasm-wrappers/fdw/langfuse_fdw/src/lib.rs @@ -0,0 +1,534 @@ +#[allow(warnings)] +mod bindings; +use serde_json::Value as JsonValue; + +use base64::{Engine, engine::general_purpose::STANDARD as BASE64}; +use bindings::{ + exports::supabase::wrappers::routines::Guest, + supabase::wrappers::{ + http, time, + types::{ + Cell, Context, FdwError, FdwResult, ImportForeignSchemaStmt, Options, OptionsType, Row, + TypeOid, Value, + }, + utils, + }, +}; + +// Langfuse returns a page of rows plus an opaque cursor. We fetch lazily: one page is +// buffered at a time and the next page is pulled once the buffer is drained, so a +// `limit 10` on a million-row project costs a single request. +#[derive(Debug, Default)] +struct LangfuseFdw { + base_url: String, + auth_header: String, + // resolved per-scan from the foreign table's `object` option + endpoint: String, + // query string carrying pushed-down filters, rebuilt per page with a new cursor + filter_qs: String, + // `fields` field-group selection for the v2 endpoints; empty for the others + fields: String, + src_rows: Vec, + src_idx: usize, + next_cursor: Option, + // page number for the offset-paginated endpoints; 0 means "first request" + next_page: i64, + // None once the API stops handing back a cursor + has_more: bool, + // rows still to return; None means unlimited + remaining: Option, + page_size: i64, + // log each outgoing request URL + verbose: bool, +} + +static mut INSTANCE: *mut LangfuseFdw = std::ptr::null_mut::(); + +impl LangfuseFdw { + fn init_instance() { + let instance = Self::default(); + unsafe { + INSTANCE = Box::leak(Box::new(instance)); + } + } + + fn this_mut() -> &'static mut Self { + unsafe { &mut (*INSTANCE) } + } + + // Resolves one credential from `_name`, `_id`, or `` in that order. + // Vault is preferred; the plaintext form is a convenience for local development. + fn read_key(opts: &Options, name: &str) -> Result { + if let Some(secret_name) = opts.get(&format!("{}_name", name)) { + return utils::get_vault_secret_by_name(&secret_name) + .ok_or(format!("secret '{}' not found in Vault", secret_name)); + } + if let Some(secret_id) = opts.get(&format!("{}_id", name)) { + return utils::get_vault_secret(&secret_id) + .ok_or(format!("secret id '{}' not found in Vault", secret_id)); + } + opts.require(name) + } + + // Only filters Langfuse accepts as query params are pushed down; everything else is + // left for Postgres to re-check locally. Postgres re-checks all of them anyway, so a + // pushdown that is merely coarse is still safe. + fn pushdown_quals(&self, ctx: &Context) -> String { + const PUSHABLE: [&str; 6] = ["trace_id", "user_id", "session_id", "type", "level", "name"]; + + // The time column and its query params differ per endpoint: observations filter + // on start_time via fromStartTime/toStartTime, while traces, sessions, and scores + // filter on timestamp via fromTimestamp/toTimestamp. + let (time_col, from_param, to_param) = if self.endpoint.ends_with("observations") { + ("start_time", "fromStartTime", "toStartTime") + } else { + ("timestamp", "fromTimestamp", "toTimestamp") + }; + + let mut qs = String::new(); + for qual in ctx.get_quals().iter() { + if qual.use_or() { + continue; + } + let field = qual.field(); + let operator = qual.operator(); + + // parameterised quals resolve too late to help us here + let Value::Cell(cell) = qual.value() else { + continue; + }; + + if field == time_col { + let micros = match cell { + Cell::Timestamp(v) | Cell::Timestamptz(v) => v, + _ => continue, + }; + // Both API bounds are inclusive-from / exclusive-to, so only the + // operators that match those semantics are pushed. `>` and `<=` would + // need an epsilon shift; Postgres filters those locally instead. + let param = match operator.as_str() { + ">=" => from_param, + "<" => to_param, + _ => continue, + }; + // Cell::Timestamp is microseconds since the epoch, and despite its name + // epoch_ms_to_rfc3339 takes microseconds too (the host calls + // from_timestamp_micros), so this passes straight through. + let Ok(iso) = time::epoch_ms_to_rfc3339(micros) else { + continue; + }; + qs.push_str(&format!("&{}={}", param, url_encode(&iso))); + continue; + } + + if operator != "=" || !PUSHABLE.contains(&field.as_str()) { + continue; + } + let Cell::String(v) = cell else { + continue; + }; + qs.push_str(&format!("&{}={}", to_camel_case(&field), url_encode(&v))); + } + qs + } + + fn fetch_page(&mut self) -> Result<(), FdwError> { + // ask for no more rows than the query still needs + let limit = match self.remaining { + Some(n) if n < self.page_size => n, + _ => self.page_size, + }; + + let mut url = format!( + "{}/api/public/{}?limit={}{}{}", + self.base_url, self.endpoint, limit, self.fields, self.filter_qs + ); + if let Some(cursor) = &self.next_cursor { + url.push_str(&format!("&cursor={}", url_encode(cursor))); + } else if self.next_page > 1 { + url.push_str(&format!("&page={}", self.next_page)); + } + + // Set `verbose 'true'` on the server to see which filters actually reached the + // API — the difference between a pushdown working and Postgres quietly filtering + // a full scan is otherwise invisible. + if self.verbose { + utils::report_info(&format!("langfuse_fdw: GET {}", url)); + } + + let headers: Vec<(String, String)> = vec![ + ("authorization".to_owned(), self.auth_header.clone()), + ("user-agent".to_owned(), "langfuse-wasm-fdw".to_owned()), + ("accept".to_owned(), "application/json".to_owned()), + ]; + + let req = http::Request { + method: http::Method::Get, + url, + headers, + body: String::default(), + }; + let resp = http::get(&req)?; + // surfaces 401/403 as a Postgres error rather than an empty result set + http::error_for_status(&resp).map_err(|err| format!("{}: {}", err, resp.body))?; + + let resp_json: JsonValue = serde_json::from_str(&resp.body).map_err(|e| e.to_string())?; + + self.src_rows = resp_json + .get("data") + .and_then(|v| v.as_array()) + .map(|v| v.to_owned()) + .ok_or("response has no 'data' array")?; + self.src_idx = 0; + + // Two pagination styles are in play: the v2/v3 endpoints hand back an opaque + // cursor, the older ones report page/totalPages. Follow whichever the response + // carries so one wrapper serves both. + let meta = resp_json.get("meta"); + self.next_cursor = meta + .and_then(|m| m.get("cursor")) + .and_then(|c| c.as_str()) + .map(|s| s.to_owned()); + + if self.next_cursor.is_some() { + self.has_more = !self.src_rows.is_empty(); + } else { + let page = meta.and_then(|m| m.get("page")).and_then(|p| p.as_i64()); + let total_pages = meta + .and_then(|m| m.get("totalPages")) + .and_then(|p| p.as_i64()); + match (page, total_pages) { + (Some(p), Some(total)) => { + self.next_page = p + 1; + self.has_more = p < total && !self.src_rows.is_empty(); + } + // no pagination metadata at all: treat as a single page + _ => self.has_more = false, + } + } + + Ok(()) + } +} + +// Postgres columns are snake_case by convention, the Langfuse API is camelCase. +fn to_camel_case(s: &str) -> String { + let mut out = String::with_capacity(s.len()); + let mut upper_next = false; + for c in s.chars() { + if c == '_' { + upper_next = true; + } else if upper_next { + out.push(c.to_ascii_uppercase()); + upper_next = false; + } else { + out.push(c); + } + } + out +} + +fn url_encode(s: &str) -> String { + let mut out = String::with_capacity(s.len()); + for b in s.as_bytes() { + match b { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => { + out.push(*b as char) + } + _ => out.push_str(&format!("%{:02X}", b)), + } + } + out +} + +// `usage_details` and `cost_details` arrive as objects keyed by metric name, with a +// `total` key summing the rest. Flattening the common totals into scalar columns keeps +// the useful aggregations (sum(total_cost)) out of jsonb extraction. +fn lookup_source<'a>(src_row: &'a JsonValue, col: &str) -> Option<&'a JsonValue> { + let obj = src_row.as_object()?; + + let direct = obj.get(&to_camel_case(col)); + if direct.is_some_and(|v| !v.is_null()) { + return direct; + } + + // Fall back to the flat aliases the older endpoints use: costs are exposed as + // `calculated*Cost` there, token counts as `promptTokens`/`completionTokens`. + let fallback = |names: &[&str]| -> Option<&'a JsonValue> { + names + .iter() + .filter_map(|n| obj.get(*n)) + .find(|v| !v.is_null()) + }; + + let (group, key) = match col { + "total_cost" => ("costDetails", "total"), + "input_cost" => ("costDetails", "input"), + "output_cost" => ("costDetails", "output"), + "total_tokens" => ("usageDetails", "total"), + "input_tokens" => ("usageDetails", "input"), + "output_tokens" => ("usageDetails", "output"), + // v2 renamed `model` to `providedModelName`; accept either spelling so the + // same foreign table definition works against both. + "provided_model_name" => return obj.get("model"), + "model" => return obj.get("providedModelName"), + _ => return direct, + }; + + obj.get(group) + .and_then(|g| g.as_object()) + .and_then(|g| g.get(key)) + .filter(|v| !v.is_null()) + .or_else(|| match col { + "total_cost" => fallback(&["calculatedTotalCost", "totalCost", "totalPrice"]), + "input_cost" => fallback(&["calculatedInputCost", "inputPrice"]), + "output_cost" => fallback(&["calculatedOutputCost", "outputPrice"]), + "input_tokens" => fallback(&["promptTokens"]), + "output_tokens" => fallback(&["completionTokens"]), + _ => None, + }) + .or(direct) +} + +impl Guest for LangfuseFdw { + fn host_version_requirement() -> String { + // semver expression for Wasm FDW host version requirement + // ref: https://docs.rs/semver/latest/semver/enum.Op.html + "^0.1.0".to_string() + } + + fn init(ctx: &Context) -> FdwResult { + Self::init_instance(); + let this = Self::this_mut(); + + let opts = ctx.get_options(&OptionsType::Server); + this.base_url = opts + .require_or("api_url", "https://cloud.langfuse.com") + .trim_end_matches('/') + .to_owned(); + this.page_size = opts + .require_or("page_size", "100".to_owned().as_str()) + .parse::() + .map_err(|e| format!("invalid page_size: {}", e))?; + this.verbose = opts.require_or("verbose", "false") == "true"; + + // Keys live in Vault so they never appear in `create server` DDL or pg_dump + // output. Accepts a secret name, a secret id, or a plaintext key — the name form + // saves having to look a UUID back up. + let public_key = Self::read_key(&opts, "public_key")?; + let secret_key = Self::read_key(&opts, "secret_key")?; + + this.auth_header = format!( + "Basic {}", + BASE64.encode(format!("{}:{}", public_key, secret_key)) + ); + + Ok(()) + } + + fn begin_scan(ctx: &Context) -> FdwResult { + let this = Self::this_mut(); + + let opts = ctx.get_options(&OptionsType::Table); + this.endpoint = opts.require("object")?; + // reads self.endpoint to pick the right time filter params, so it must run after + this.filter_qs = this.pushdown_quals(ctx); + + // The v2 endpoints return only the `core` and `basic` groups unless asked + // otherwise, which would leave usage/cost columns silently NULL. Request the + // groups backing the columns this wrapper exposes. + this.fields = match opts.get("fields") { + Some(f) => format!("&fields={}", url_encode(&f)), + None if this.endpoint.starts_with("v2/") => { + "&fields=core,basic,time,io,metadata,model,usage,metrics,trace_context".to_owned() + } + None => String::new(), + }; + + // A pushed-down LIMIT bounds total rows fetched. Postgres re-checks it anyway, + // so over-fetching would only waste requests. + this.remaining = ctx + .get_limit() + .map(|l| l.offset().saturating_add(l.count())); + + this.next_cursor = None; + this.has_more = false; + this.fetch_page() + } + + fn iter_scan(ctx: &Context, row: &Row) -> Result, FdwError> { + let this = Self::this_mut(); + + if this.remaining == Some(0) { + return Ok(None); + } + + // current page drained: pull the next one, if any + if this.src_idx >= this.src_rows.len() { + if !this.has_more { + return Ok(None); + } + this.fetch_page()?; + if this.src_rows.is_empty() { + return Ok(None); + } + } + + let src_row = &this.src_rows[this.src_idx]; + for tgt_col in ctx.get_columns() { + let tgt_col_name = tgt_col.name(); + // A column absent from this field group is NULL rather than an error — + // `fields` selection means the API legitimately omits keys. + let src = match lookup_source(src_row, &tgt_col_name) { + Some(v) => v, + None => { + row.push(None); + continue; + } + }; + let cell = match tgt_col.type_oid() { + TypeOid::Bool => src.as_bool().map(Cell::Bool), + TypeOid::String => match src { + // input/output/metadata are arbitrary JSON; render them as text + // instead of failing when a text column receives an object + JsonValue::String(s) => Some(Cell::String(s.to_owned())), + JsonValue::Null => None, + other => Some(Cell::String(other.to_string())), + }, + TypeOid::I32 => src.as_i64().map(|v| Cell::I32(v as i32)), + TypeOid::I64 => src.as_i64().map(Cell::I64), + TypeOid::F64 => src.as_f64().map(Cell::F64), + TypeOid::Numeric => src.as_f64().map(Cell::Numeric), + TypeOid::Timestamp => match src.as_str() { + Some(s) => Some(Cell::Timestamp(time::parse_from_rfc3339(s)?)), + None => None, + }, + TypeOid::Timestamptz => match src.as_str() { + Some(s) => Some(Cell::Timestamptz(time::parse_from_rfc3339(s)?)), + None => None, + }, + TypeOid::Json => match src { + JsonValue::Null => None, + other => Some(Cell::Json(other.to_string())), + }, + _ => { + return Err(format!( + "column {} data type is not supported", + tgt_col_name + )); + } + }; + + row.push(cell.as_ref()); + } + + this.src_idx += 1; + this.remaining = this.remaining.map(|n| n - 1); + + Ok(Some(0)) + } + + fn re_scan(_ctx: &Context) -> FdwResult { + Err("re_scan on foreign table is not supported".to_owned()) + } + + fn end_scan(_ctx: &Context) -> FdwResult { + let this = Self::this_mut(); + this.src_rows.clear(); + this.src_idx = 0; + this.next_cursor = None; + this.has_more = false; + Ok(()) + } + + fn begin_modify(_ctx: &Context) -> FdwResult { + Err("modify on foreign table is not supported".to_owned()) + } + + fn insert(_ctx: &Context, _row: &Row) -> FdwResult { + Ok(()) + } + + fn update(_ctx: &Context, _rowid: Cell, _row: &Row) -> FdwResult { + Ok(()) + } + + fn delete(_ctx: &Context, _rowid: Cell) -> FdwResult { + Ok(()) + } + + fn end_modify(_ctx: &Context) -> FdwResult { + Ok(()) + } + + fn import_foreign_schema( + _ctx: &Context, + stmt: ImportForeignSchemaStmt, + ) -> Result, FdwError> { + Ok(vec![ + // Traces carry user_id and an aggregate cost, so this is the table to join + // local user tables against. + format!( + r#"create foreign table if not exists traces ( + id text, + name text, + user_id text, + session_id text, + environment text, + release text, + version text, + total_cost double precision, + latency double precision, + timestamp timestamp, + created_at timestamp, + updated_at timestamp, + input text, + output text, + metadata jsonb, + tags jsonb + ) + server {} options ( + object 'traces', + rowid_column 'id' + )"#, + stmt.server_name, + ), + // One row per model call, with token counts and per-call cost. No user_id in + // the response — that lives on the trace; join through trace_id. + format!( + r#"create foreign table if not exists observations ( + id text, + trace_id text, + type text, + name text, + level text, + model text, + input_tokens bigint, + output_tokens bigint, + total_tokens bigint, + input_cost double precision, + output_cost double precision, + total_cost double precision, + latency double precision, + time_to_first_token double precision, + start_time timestamp, + end_time timestamp, + completion_start_time timestamp, + prompt_name text, + prompt_version bigint, + input text, + output text, + metadata jsonb, + model_parameters jsonb + ) + server {} options ( + object 'observations', + rowid_column 'id' + )"#, + stmt.server_name, + ), + ]) + } +} + +bindings::export!(LangfuseFdw with_types_in bindings); diff --git a/wasm-wrappers/fdw/langfuse_fdw/wit/world.wit b/wasm-wrappers/fdw/langfuse_fdw/wit/world.wit new file mode 100644 index 000000000..cc2607aaa --- /dev/null +++ b/wasm-wrappers/fdw/langfuse_fdw/wit/world.wit @@ -0,0 +1,10 @@ +package supabase:langfuse-fdw@0.1.0; + +world langfuse { + import supabase:wrappers/http@0.2.0; + import supabase:wrappers/jwt@0.2.0; + import supabase:wrappers/stats@0.2.0; + import supabase:wrappers/time@0.2.0; + import supabase:wrappers/utils@0.2.0; + export supabase:wrappers/routines@0.2.0; +} From a9a09c1a224f3af8b2709247dece90fa4db53699 Mon Sep 17 00:00:00 2001 From: mia <35839923+distanceqo@users.noreply.github.com> Date: Tue, 11 Aug 2026 20:36:39 +0800 Subject: [PATCH 2/5] Inline format args to satisfy clippy CI runs clippy with RUSTFLAGS="-D warnings", which promotes uninlined_format_args to an error. Nine call sites used the positional form. --- wasm-wrappers/fdw/langfuse_fdw/src/lib.rs | 21 +++++++++------------ 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/wasm-wrappers/fdw/langfuse_fdw/src/lib.rs b/wasm-wrappers/fdw/langfuse_fdw/src/lib.rs index ac5c34ef6..2d7f77ac2 100644 --- a/wasm-wrappers/fdw/langfuse_fdw/src/lib.rs +++ b/wasm-wrappers/fdw/langfuse_fdw/src/lib.rs @@ -59,13 +59,13 @@ impl LangfuseFdw { // Resolves one credential from `_name`, `_id`, or `` in that order. // Vault is preferred; the plaintext form is a convenience for local development. fn read_key(opts: &Options, name: &str) -> Result { - if let Some(secret_name) = opts.get(&format!("{}_name", name)) { + if let Some(secret_name) = opts.get(&format!("{name}_name")) { return utils::get_vault_secret_by_name(&secret_name) - .ok_or(format!("secret '{}' not found in Vault", secret_name)); + .ok_or(format!("secret '{secret_name}' not found in Vault")); } - if let Some(secret_id) = opts.get(&format!("{}_id", name)) { + if let Some(secret_id) = opts.get(&format!("{name}_id")) { return utils::get_vault_secret(&secret_id) - .ok_or(format!("secret id '{}' not found in Vault", secret_id)); + .ok_or(format!("secret id '{secret_id}' not found in Vault")); } opts.require(name) } @@ -153,7 +153,7 @@ impl LangfuseFdw { // API — the difference between a pushdown working and Postgres quietly filtering // a full scan is otherwise invisible. if self.verbose { - utils::report_info(&format!("langfuse_fdw: GET {}", url)); + utils::report_info(&format!("langfuse_fdw: GET {url}")); } let headers: Vec<(String, String)> = vec![ @@ -235,7 +235,7 @@ fn url_encode(s: &str) -> String { b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => { out.push(*b as char) } - _ => out.push_str(&format!("%{:02X}", b)), + _ => out.push_str(&format!("%{b:02X}")), } } out @@ -309,7 +309,7 @@ impl Guest for LangfuseFdw { this.page_size = opts .require_or("page_size", "100".to_owned().as_str()) .parse::() - .map_err(|e| format!("invalid page_size: {}", e))?; + .map_err(|e| format!("invalid page_size: {e}"))?; this.verbose = opts.require_or("verbose", "false") == "true"; // Keys live in Vault so they never appear in `create server` DDL or pg_dump @@ -320,7 +320,7 @@ impl Guest for LangfuseFdw { this.auth_header = format!( "Basic {}", - BASE64.encode(format!("{}:{}", public_key, secret_key)) + BASE64.encode(format!("{public_key}:{secret_key}")) ); Ok(()) @@ -412,10 +412,7 @@ impl Guest for LangfuseFdw { other => Some(Cell::Json(other.to_string())), }, _ => { - return Err(format!( - "column {} data type is not supported", - tgt_col_name - )); + return Err(format!("column {tgt_col_name} data type is not supported")); } }; From 91bbc141600cec4ccfd1312751a009f8e611821d Mon Sep 17 00:00:00 2001 From: mia <35839923+distanceqo@users.noreply.github.com> Date: Wed, 12 Aug 2026 20:42:27 +0800 Subject: [PATCH 3/5] Add smoke test for the Langfuse Wasm FDW MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mock responses are shaped so a regression fails rather than silently passing: - Both endpoints report page/totalPages instead of a cursor, and the server is configured with page_size 1 against a two-page traces mock, so the test only sees both rows if pagination is followed. - The traces mock returns nothing for any userId other than user-alice, so a dropped equality pushdown fails the assertion instead of falling back to unfiltered rows. - The two observation rows spell usage and cost differently — one nests them under usageDetails/costDetails, the other only has the flat promptTokens and calculated*Cost forms. Both must map to the same columns, so a missing fallback surfaces as a NULL. --- wrappers/dockerfiles/wasm/server.py | 99 +++++++++++++++++++++++++ wrappers/src/fdw/wasm_fdw/tests.rs | 110 ++++++++++++++++++++++++++++ 2 files changed, 209 insertions(+) diff --git a/wrappers/dockerfiles/wasm/server.py b/wrappers/dockerfiles/wasm/server.py index 698e8c49a..ae9031238 100644 --- a/wrappers/dockerfiles/wasm/server.py +++ b/wrappers/dockerfiles/wasm/server.py @@ -486,6 +486,105 @@ def do_GET(self): } } """ + elif fdw == "langfuse": + # Both endpoints report page/totalPages rather than a cursor, which is what + # the older Langfuse read APIs do. `page=1` of `totalPages=2` means a + # regression that stops following pages would return only the first row. + qs = parse_qs(urlparse(req_path).query) + page = int(qs.get("page", ["1"])[0]) + + if req_path.startswith("/api/public/traces"): + # A userId filter is pushed down as a query param. Returning nothing for + # a non-matching value keeps the pushdown assertion honest: were the + # filter dropped, the unfiltered rows below would still come back. + if qs.get("userId", [None])[0] not in (None, "user-alice"): + body = json.dumps( + {"data": [], "meta": {"page": 1, "totalPages": 1}} + ) + elif page == 1: + body = """ +{ + "data": [ + { + "id": "trace-1", + "name": "summarize-doc", + "userId": "user-alice", + "sessionId": "session-0", + "environment": "default", + "totalCost": 0.0435, + "latency": 4.2, + "timestamp": "2026-08-09T02:46:41.286Z", + "tags": ["seed"], + "metadata": {"seeded": true} + } + ], + "meta": {"page": 1, "limit": 1, "totalItems": 2, "totalPages": 2} +} + """ + else: + body = """ +{ + "data": [ + { + "id": "trace-2", + "name": "chat-turn", + "userId": "user-alice", + "sessionId": "session-1", + "environment": "default", + "totalCost": 0.0196, + "latency": 6.1, + "timestamp": "2026-08-09T01:11:41.286Z", + "tags": ["seed"], + "metadata": {"seeded": true} + } + ], + "meta": {"page": 2, "limit": 1, "totalItems": 2, "totalPages": 2} +} + """ + else: + # Deliberately mixes the two shapes the API uses for usage and cost: + # the first row nests them under usageDetails/costDetails, the second + # only has the flat calculated*/promptTokens spellings. Both must map to + # the same columns, so a missing fallback shows up as a NULL here. + body = """ +{ + "data": [ + { + "id": "obs-1", + "traceId": "trace-1", + "type": "GENERATION", + "name": "summarize-doc", + "level": "DEFAULT", + "model": "claude-opus-5", + "usageDetails": {"input": 1200, "output": 340, "total": 1540}, + "costDetails": {"input": 0.018, "output": 0.0255, "total": 0.0435}, + "latency": 4.2, + "startTime": "2026-08-09T02:46:41.286Z", + "endTime": "2026-08-09T02:46:45.486Z", + "metadata": {"seeded": true} + }, + { + "id": "obs-2", + "traceId": "trace-2", + "type": "GENERATION", + "name": "chat-turn", + "level": "DEFAULT", + "model": "claude-sonnet-5", + "promptTokens": 4300, + "completionTokens": 1100, + "totalTokens": 5400, + "calculatedInputCost": 0.0129, + "calculatedOutputCost": 0.0165, + "calculatedTotalCost": 0.0294, + "latency": 6.1, + "startTime": "2026-08-09T01:11:41.286Z", + "endTime": "2026-08-09T01:11:47.386Z", + "metadata": {"seeded": true} + } + ], + "meta": {"page": 1, "limit": 50, "totalItems": 2, "totalPages": 1} +} + """ elif fdw == "openapi": # Generic OpenAPI FDW test endpoints covering all features diff --git a/wrappers/src/fdw/wasm_fdw/tests.rs b/wrappers/src/fdw/wasm_fdw/tests.rs index 763028b6a..f7dde7a36 100644 --- a/wrappers/src/fdw/wasm_fdw/tests.rs +++ b/wrappers/src/fdw/wasm_fdw/tests.rs @@ -917,6 +917,116 @@ mod tests { .filter_map(|r| r.get_by_name::<&str, _>("received_auth").unwrap()) .collect::>(); assert_eq!(injected, vec!["Bearer sess-secret-123"]); + + // Langfuse FDW test + c.update( + r#"CREATE SERVER langfuse_server + FOREIGN DATA WRAPPER wasm_wrapper + OPTIONS ( + fdw_package_url 'file://../../../wasm-wrappers/fdw/target/wasm32-unknown-unknown/release/langfuse_fdw.wasm', + fdw_package_name 'supabase:langfuse-fdw', + fdw_package_version '>=0.1.0', + api_url 'http://localhost:8096/langfuse', + public_key 'pk-lf-aaa', + secret_key 'sk-lf-bbb', + page_size '1' + )"#, + None, + &[], + ) + .unwrap(); + c.update( + r#" + CREATE FOREIGN TABLE langfuse_traces ( + id text, + name text, + user_id text, + total_cost double precision, + latency double precision, + timestamp timestamp, + metadata jsonb + ) + SERVER langfuse_server + OPTIONS ( + object 'traces', + rowid_column 'id' + ) + "#, + None, + &[], + ) + .unwrap(); + + // page_size is 1 against a two-page mock, so this only returns both rows if + // pagination is followed past the first response. + let results = c + .select("SELECT * FROM langfuse_traces ORDER BY id", None, &[]) + .unwrap() + .filter_map(|r| r.get_by_name::<&str, _>("id").unwrap()) + .collect::>(); + assert_eq!(results, vec!["trace-1", "trace-2"]); + + // Equality on user_id is pushed down as a query param. The mock answers with + // no rows for any other value, so a dropped filter fails here. + let filtered = c + .select( + "SELECT id FROM langfuse_traces WHERE user_id = 'user-bob'", + None, + &[], + ) + .unwrap() + .filter_map(|r| r.get_by_name::<&str, _>("id").unwrap()) + .collect::>(); + assert!(filtered.is_empty()); + + c.update( + r#" + CREATE FOREIGN TABLE langfuse_observations ( + id text, + trace_id text, + model text, + input_tokens bigint, + output_tokens bigint, + total_tokens bigint, + total_cost double precision, + start_time timestamp + ) + SERVER langfuse_server + OPTIONS ( + object 'observations', + rowid_column 'id' + ) + "#, + None, + &[], + ) + .unwrap(); + + // The two mock rows spell usage and cost differently: obs-1 nests them under + // usageDetails/costDetails, obs-2 uses the flat promptTokens and + // calculatedTotalCost forms. Both must land in the same columns, so a missing + // fallback shows up as a NULL that filter_map drops. + let tokens = c + .select( + "SELECT input_tokens FROM langfuse_observations ORDER BY id", + None, + &[], + ) + .unwrap() + .filter_map(|r| r.get_by_name::("input_tokens").unwrap()) + .collect::>(); + assert_eq!(tokens, vec![1200, 4300]); + + let costs = c + .select( + "SELECT total_cost FROM langfuse_observations ORDER BY id", + None, + &[], + ) + .unwrap() + .filter_map(|r| r.get_by_name::("total_cost").unwrap()) + .collect::>(); + assert_eq!(costs, vec![0.0435, 0.0294]); }); } } From 62b94910c3c2fff39f9a09a02dd0fefac1cda916 Mon Sep 17 00:00:00 2001 From: Bo Lu Date: Fri, 14 Aug 2026 17:49:09 +1000 Subject: [PATCH 4/5] fix(langfuse_fdw): validate page_size against documented 1-1000 range Previously page_size was only checked for being a valid integer; values outside the documented 1-1000 range (e.g. negative or larger than 1000) were silently accepted and passed straight through as the upstream limit= query parameter. Co-Authored-By: Claude Sonnet 5 --- wasm-wrappers/fdw/langfuse_fdw/src/lib.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/wasm-wrappers/fdw/langfuse_fdw/src/lib.rs b/wasm-wrappers/fdw/langfuse_fdw/src/lib.rs index 2d7f77ac2..043d0e3fd 100644 --- a/wasm-wrappers/fdw/langfuse_fdw/src/lib.rs +++ b/wasm-wrappers/fdw/langfuse_fdw/src/lib.rs @@ -310,6 +310,9 @@ impl Guest for LangfuseFdw { .require_or("page_size", "100".to_owned().as_str()) .parse::() .map_err(|e| format!("invalid page_size: {e}"))?; + if !(1..=1000).contains(&this.page_size) { + return Err("invalid page_size: must be between 1 and 1000".to_owned()); + } this.verbose = opts.require_or("verbose", "false") == "true"; // Keys live in Vault so they never appear in `create server` DDL or pg_dump From 6b492a7337d46500a49787831e428114a4f1e04f Mon Sep 17 00:00:00 2001 From: Bo Lu Date: Fri, 14 Aug 2026 18:55:30 +1000 Subject: [PATCH 5/5] update README and author --- README.md | 1 + docs/catalog/langfuse.md | 4 ++-- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 4db83c790..d0637b6d8 100644 --- a/README.md +++ b/README.md @@ -22,6 +22,7 @@ | [HelloWorld](./wrappers/src/fdw/helloworld_fdw) | A demo FDW to show how to develop a basic FDW. | | | | [HubSpot](./wasm-wrappers/fdw/hubspot_fdw) | A Wasm FDW for [HubSpot](https://www.hubspot.com/) | ✅ | ❌ | | [Infura](./wasm-wrappers/fdw/infura_fdw) | A Wasm FDW for [Infura](https://www.infura.io/) blockchain data | ✅ | ❌ | +| [Langfuse](./wasm-wrappers/fdw/langfuse_fdw) | A Wasm FDW for [Langfuse](https://langfuse.com/) | ✅ | ❌ | | [Logflare](./wrappers/src/fdw/logflare_fdw) | A FDW for [Logflare](https://logflare.app/) | ✅ | ❌ | | [MongoDB](./wrappers/src/fdw/mongodb_fdw) | A FDW for [MongoDB](https://www.mongodb.com/) | ✅ | ✅ | | [MySQL](./wrappers/src/fdw/mysql_fdw) | A FDW for [MySQL](https://www.mysql.com/) | ✅ | ✅ | diff --git a/docs/catalog/langfuse.md b/docs/catalog/langfuse.md index fde6f4c54..d07add626 100644 --- a/docs/catalog/langfuse.md +++ b/docs/catalog/langfuse.md @@ -1,10 +1,10 @@ --- source: documentation: -author: supabase +author: distanceqo(https://github.com/distanceqo) tags: - wasm - - official + - community --- # Langfuse