From ecf5d4babd4eda45e85b59e55678e21be120aafa Mon Sep 17 00:00:00 2001 From: Yudhi Armyndharis Date: Thu, 25 Jun 2026 22:44:58 +0700 Subject: [PATCH] feat(voice-transcription): speech-to-text plugin for inbound voice notes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the voice-transcription marketplace extension: transcribes inbound WhatsApp voice notes via an OpenAI-compatible STT backend (self-hosted Speaches/faster-whisper, or hosted Groq/OpenAI) and delivers a message.transcription event out-of-band, so bots and AI can read and reply to audio. Implements the request in rmyndharis/OpenWA#365. - Off the message-delivery critical path: the message:received hook returns immediately and STT runs as an un-awaited task, so it never blocks or delays delivery (and is not bound by the 5s hook budget). - Audio uploaded as a binary multipart Buffer body (intact across the sandbox boundary); part labeled voice.ogg so OGG/Opus needs no transcode. - Delivery: configurable webhook (HMAC-SHA256 signed in X-OpenWA-Signature, matching core webhooks) and/or optional in-chat (off|self|reply, default off; self avoids leaking to the sender). Either is optional. - Status events: completed / failed / skipped(reason). - Guards: exact maxSizeBytes, per-session hourly rate limit, best-effort idempotency (suppresses #466-style engine re-fires), STT circuit breaker. Fail-open throughout. Contract: widen the vendored types to match the sandbox runtime — PluginNetResponse.body (the real field; the .json()/.text() methods do not cross the worker boundary) and IncomingMessage.media. Also updates the group-translate test fixture for the now-required body field. --- README.md | 1 + group-translate/libretranslate.client.test.ts | 1 + package.json | 2 +- plugins.json | 33 +++ tsconfig.json | 1 + types/openwa.d.ts | 22 +- voice-transcription/CHANGELOG.md | 35 +++ voice-transcription/README.md | 157 +++++++++++++ voice-transcription/index.test.ts | 91 ++++++++ voice-transcription/index.ts | 120 ++++++++++ voice-transcription/manifest.json | 72 ++++++ voice-transcription/multipart.test.ts | 52 +++++ voice-transcription/multipart.ts | 40 ++++ voice-transcription/openai-stt.client.test.ts | 142 ++++++++++++ voice-transcription/openai-stt.client.ts | 106 +++++++++ .../transcription.coordinator.test.ts | 207 ++++++++++++++++++ .../transcription.coordinator.ts | 158 +++++++++++++ voice-transcription/webhook.delivery.test.ts | 72 ++++++ voice-transcription/webhook.delivery.ts | 54 +++++ 19 files changed, 1364 insertions(+), 2 deletions(-) create mode 100644 voice-transcription/CHANGELOG.md create mode 100644 voice-transcription/README.md create mode 100644 voice-transcription/index.test.ts create mode 100644 voice-transcription/index.ts create mode 100644 voice-transcription/manifest.json create mode 100644 voice-transcription/multipart.test.ts create mode 100644 voice-transcription/multipart.ts create mode 100644 voice-transcription/openai-stt.client.test.ts create mode 100644 voice-transcription/openai-stt.client.ts create mode 100644 voice-transcription/transcription.coordinator.test.ts create mode 100644 voice-transcription/transcription.coordinator.ts create mode 100644 voice-transcription/webhook.delivery.test.ts create mode 100644 voice-transcription/webhook.delivery.ts diff --git a/README.md b/README.md index f9d9f95..f74cde8 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,7 @@ This repository provides: | [`faq-bot`](./faq-bot) | Auto-replies to inbound WhatsApp messages from configurable FAQ keyword/regex rules. | 0.1.4 | stable | | [`group-translate`](./group-translate) | Auto-translates group messages between participants' languages via a LibreTranslate backend. Configure in-chat with /tr commands. Admin-gated; disabled until enabled. | 1.0.3 | stable | | [`gsheets-logger`](./gsheets-logger) | Logs WhatsApp message events to a Google Sheet via a service account. | 0.2.2 | stable | +| [`voice-transcription`](./voice-transcription) | Transcribes inbound WhatsApp voice notes to text via an OpenAI-compatible speech-to-text backend (self-hosted Speaches/faster-whisper or hosted Groq/OpenAI) and delivers a `message.transcription` event to your webhook — so bots and AI can read and reply to audio. Off the message-delivery path; disabled until enabled. | 1.0.0 | beta | The table above is generated from each plugin's `manifest.json` + `CHANGELOG.md` by `npm run catalog` diff --git a/group-translate/libretranslate.client.test.ts b/group-translate/libretranslate.client.test.ts index b4c942a..0d81c77 100644 --- a/group-translate/libretranslate.client.test.ts +++ b/group-translate/libretranslate.client.test.ts @@ -8,6 +8,7 @@ function res(partial: { ok?: boolean; status?: number; json?: () => Promise '', // PluginNetResponse.json is generic (() => Promise); a concrete fake needs the cast. json: (partial.json ?? (async () => ({}))) as PluginNetResponse['json'], diff --git a/package.json b/package.json index 196cb83..d162bee 100644 --- a/package.json +++ b/package.json @@ -9,7 +9,7 @@ "build": "node package.mjs gsheets-logger", "catalog": "node scripts/catalog.mjs", "catalog:check": "node scripts/catalog.mjs --check", - "test": "node --import tsx --test \"gsheets-logger/*.test.ts\" \"faq-bot/*.test.ts\" \"after-hours/*.test.ts\" \"chat-flow/*.test.ts\" \"group-translate/**/*.test.ts\"", + "test": "node --import tsx --test \"gsheets-logger/*.test.ts\" \"faq-bot/*.test.ts\" \"after-hours/*.test.ts\" \"chat-flow/*.test.ts\" \"group-translate/**/*.test.ts\" \"voice-transcription/**/*.test.ts\"", "typecheck": "tsc --noEmit" }, "devDependencies": { diff --git a/plugins.json b/plugins.json index 15b2a0a..d9ad249 100644 --- a/plugins.json +++ b/plugins.json @@ -974,5 +974,38 @@ } } } + }, + { + "id": "voice-transcription", + "name": "Voice Note Transcription", + "version": "1.0.0", + "type": "extension", + "status": "beta", + "description": "Transcribes inbound WhatsApp voice notes to text via an OpenAI-compatible speech-to-text backend (self-hosted Speaches/faster-whisper or hosted Groq/OpenAI) and delivers a `message.transcription` event to your webhook — so bots and AI can read and reply to audio. Off the message-delivery path; disabled until enabled.", + "author": "Yudhi Armyndharis ", + "license": "MIT", + "keywords": [ + "transcription", + "speech-to-text", + "stt", + "whisper", + "voice", + "audio", + "whatsapp", + "openwa" + ], + "minOpenWAVersion": "0.7.0", + "testedOpenWAVersion": "0.7.3", + "releasedAt": "2026-06-25", + "repoPath": "voice-transcription", + "repoUrl": "https://github.com/rmyndharis/OpenWA-plugins", + "homepage": "https://github.com/rmyndharis/OpenWA-plugins/tree/main/voice-transcription", + "download": "https://github.com/rmyndharis/OpenWA-plugins/releases/download/voice-transcription-v1.0.0/voice-transcription.zip", + "i18n": { + "es": { + "name": "Transcripción de Notas de Voz", + "description": "Transcribe las notas de voz entrantes de WhatsApp a texto mediante un backend de voz a texto compatible con OpenAI (Speaches/faster-whisper autoalojado o Groq/OpenAI) y entrega un evento message.transcription a tu webhook, para que los bots y la IA puedan leer y responder al audio. Fuera de la ruta de entrega de mensajes; desactivado hasta que se habilite." + } + } } ] diff --git a/tsconfig.json b/tsconfig.json index c5ee4cb..11c7527 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -15,6 +15,7 @@ "after-hours/**/*.ts", "chat-flow/**/*.ts", "group-translate/**/*.ts", + "voice-transcription/**/*.ts", "types/**/*.d.ts" ] } diff --git a/types/openwa.d.ts b/types/openwa.d.ts index 3271b1b..c5b942b 100644 --- a/types/openwa.d.ts +++ b/types/openwa.d.ts @@ -68,14 +68,24 @@ export interface PluginEngineReadCapability { export interface PluginNetRequestInit { method?: string; headers?: Record; - body?: string; + // The sandbox bridges the request to the host via structuredClone, which preserves typed arrays — + // so a binary body (e.g. an assembled multipart/form-data upload) is sent intact. A string body is + // UTF-8 encoded by the host fetch, so binary MUST be passed as Uint8Array/Buffer, not a string. + body?: string | Uint8Array; timeoutMs?: number; } export interface PluginNetResponse { ok: boolean; status: number; + statusText?: string; headers: Record; + // The actual field the sandbox runtime returns: the response body, read host-side (capped at 10 MiB) + // and handed back as a UTF-8 string. Parse JSON with `JSON.parse(res.body)`. + body: string; + // NOTE: these method forms are NOT provided by the sandbox runtime (functions cannot cross the + // worker structuredClone boundary). Use `body` above; the methods are retained only so older + // plugins still type-check. Calling them at runtime throws. text(): Promise; json(): Promise; arrayBuffer(): Promise; @@ -190,4 +200,14 @@ export interface IncomingMessage { senderPhone?: string | null; mentionedIds?: string[]; contact?: { name?: string; pushName?: string }; + // Inbound media, materialized by the adapter before the hook fires (both engines). `data` is base64 + // and ABSENT when `omitted` is true (the payload exceeded the inbound size cap; `sizeBytes` is still + // set). For a voice note `type` is `'voice'` and `mimetype` is typically `'audio/ogg; codecs=opus'`. + media?: { + mimetype: string; + filename?: string; + data?: string; + omitted?: boolean; + sizeBytes?: number; + }; } diff --git a/voice-transcription/CHANGELOG.md b/voice-transcription/CHANGELOG.md new file mode 100644 index 0000000..37bc137 --- /dev/null +++ b/voice-transcription/CHANGELOG.md @@ -0,0 +1,35 @@ +# Changelog + +All notable changes to the Voice Note Transcription plugin are documented here. The format is based on +[Keep a Changelog](https://keepachangelog.com/en/1.0.0/) and this project adheres to +[Semantic Versioning](https://semver.org/). + +## [Unreleased] + +## [1.0.0] — 2026-06-25 + +### Added + +- Initial release. Transcribes inbound WhatsApp voice notes via an OpenAI-compatible + `/v1/audio/transcriptions` backend (self-hosted Speaches/faster-whisper, or hosted Groq/OpenAI) and + delivers a `message.transcription` event to a configurable webhook — the integration channel for + bots/AI to read and reply to audio. +- Runs **off the message-delivery critical path**: the `message:received` hook returns immediately and + the STT call + delivery run as an un-awaited promise, so transcription never blocks or delays message + delivery (and is not bound by the host's 5s hook budget). +- Audio is uploaded as a binary multipart body (intact across the sandbox boundary); the part is labeled + `voice.ogg`/`audio/ogg` so OpenAI-compatible servers accept WhatsApp's OGG/Opus without transcoding. +- Guards: message-type filter (default `voice`), exact `maxSizeBytes` cost guard, best-effort per-session + hourly rate limit, and a best-effort idempotency guard that suppresses near-simultaneous engine re-fires. +- Status events: delivers `completed` (with transcript), `failed` (STT errored), or `skipped` (too large, + rate-limited, empty) — so a consumer always knows a voice note was received even when it can't be read. +- Optional **in-chat delivery** (`chatDelivery`: `off` | `self` | `reply`, default `off`) for operators who + want the transcript inside WhatsApp; `self` notes it to your own number without leaking to the sender. + Webhook delivery is optional too — the plugin can run chat-only. +- Webhook payloads are **HMAC-SHA256 signed** in `X-OpenWA-Signature` (same scheme as OpenWA core webhooks) + when a delivery secret is set, so existing verification reuses the same check. +- STT **circuit breaker**: after repeated failures the backend is skipped for a cooldown, so a degraded + provider isn't hammered. +- Fail-open throughout — any STT or delivery error is logged and skipped, never disrupting delivery. +- The delivered transcript is marked `untrusted: true` (`source: "speech-to-text"`): downstream LLM + consumers must treat it as user-role input. diff --git a/voice-transcription/README.md b/voice-transcription/README.md new file mode 100644 index 0000000..7c954cd --- /dev/null +++ b/voice-transcription/README.md @@ -0,0 +1,157 @@ +# Voice Note Transcription + +> Transcribes inbound WhatsApp voice notes to text via an OpenAI-compatible speech-to-text backend and +> delivers a `message.transcription` event to your webhook — so bots and AI can read and reply to audio. + +![type: extension](https://img.shields.io/badge/type-extension-blue.svg) +![license: MIT](https://img.shields.io/badge/license-MIT-green.svg) +![built for OpenWA](https://img.shields.io/badge/OpenWA-%E2%89%A5%200.7.0-25D366.svg) + +## Details + + +| Field | Value | +| ----- | ----- | +| **Identifier** | `voice-transcription` | +| **Version** | 1.0.0 | +| **Released** | 2026-06-25 | +| **Status** | beta | +| **Author** | Yudhi Armyndharis | +| **License** | MIT | +| **Type** | `extension` | +| **Requires OpenWA** | ≥ 0.7.0 (tested 0.7.3) | +| **Keywords** | transcription, speech-to-text, stt, whisper, voice, audio, whatsapp, openwa | +| **Repository** | [OpenWA-plugins/voice-transcription](https://github.com/rmyndharis/OpenWA-plugins/tree/main/voice-transcription) | + + +## Features + +- **Voice → text, out of band.** On each inbound voice note the plugin runs speech-to-text and POSTs a + `message.transcription` event to your webhook. It never touches the `message.received` payload and never + replies into the contact's chat, so the transcript reaches your bot/AI without polluting the conversation + or leaking back to the sender. +- **Never blocks delivery.** The hook returns immediately; STT runs as an un-awaited task. A slow or long + transcription cannot delay (or drop) WhatsApp message delivery, and it is not bound by OpenWA's 5-second + plugin-hook budget. +- **Bring your own STT.** Any OpenAI-compatible `/v1/audio/transcriptions` endpoint: self-hosted + [Speaches](https://github.com/speaches-ai/speaches)/faster-whisper (free, local, decodes WhatsApp + OGG/Opus natively — no transcoding), or hosted Groq / OpenAI by changing one URL. +- **Status events.** Delivers `completed` (with the transcript), `failed` (STT errored), or `skipped` + (too large / rate-limited / empty) — a consumer always knows a voice note arrived, even when it can't be read. +- **Optional in-chat delivery.** `chatDelivery` can also post the transcript into WhatsApp (`self` notes it to + your own number; `reply` quote-replies to the sender). Off by default; webhook delivery is optional too + (chat-only is supported). +- **Signed webhooks.** When a delivery secret is set, the body is HMAC-SHA256 signed in `X-OpenWA-Signature` + (same scheme as OpenWA's core webhooks), so existing verification reuses the same check. +- **Cost & abuse guards.** Exact `maxSizeBytes` skip, a best-effort per-session hourly cap, a message-type + filter (voice only by default), a best-effort idempotency guard against engine re-fires, and an STT + circuit breaker that backs off a degraded backend. +- **Fail-open & guarded HTTP.** Any error is logged and skipped. All outbound calls go through the host's + SSRF-guarded `ctx.net.fetch`; the STT and delivery hosts must be allow-listed (see **Security**). The STT + and delivery secrets are stored redacted. + +## What it does + +For every inbound message whose `type` is in **Message types to transcribe** (default `voice`) and that +carries inline audio, the plugin: decodes the audio, skips it if it was dropped over the inbound size cap +or exceeds `maxSizeBytes`, applies the idempotency + hourly-rate guards, calls your STT endpoint, and — if a +non-empty transcript comes back — POSTs this to your **Delivery webhook URL**: + +```json +{ + "event": "message.transcription", + "sessionId": "…", + "messageId": "", + "chatId": "…@s.whatsapp.net", + "status": "completed", + "source": "speech-to-text", + "untrusted": true, + "transcription": { "text": "…", "language": "es", "provider": "faster-whisper", "model": "small" } +} +``` + +Correlate it to the original voice note by `messageId` (it arrives shortly **after** `message.received`, +out of order — do not assume ordering). + +## Setup + +1. **Run an STT backend.** Easiest local option — Speaches (faster-whisper), which exposes an + OpenAI-compatible API and transcribes OGG/Opus directly: + ```bash + docker run -d --name speaches -p 8000:8000 ghcr.io/speaches-ai/speaches:latest-cpu + ``` + Preload/keep a small model warm for low latency. Or use hosted Groq (`https://api.groq.com/openai`, + model `whisper-large-v3-turbo`) / OpenAI with an API key. +2. **Allow the hosts.** This plugin ships `net.allow` for `localhost`, `127.0.0.1`, `api.groq.com:443`, + `api.openai.com:443`. For any **other** STT host or **delivery webhook** host, add `host:port` to + `net.allow` in `manifest.json` and re-package (`node package.mjs voice-transcription`). +3. **For a localhost STT/delivery target**, also set `SSRF_ALLOWED_HOSTS` on the OpenWA host (the SSRF guard + blocks loopback by default), e.g. `SSRF_ALLOWED_HOSTS=127.0.0.1,localhost`. Prefer a literal + `http://127.0.0.1:PORT` target (nothing to DNS-rebind). +4. **Stand up your delivery endpoint** (an n8n webhook, a Worker, your bot) and put its URL in + **Delivery webhook URL**. + +## Install + +```bash +# Upload the packaged zip +curl -X POST http://localhost:2785/plugins/install \ + -H "X-API-Key: $OPENWA_API_KEY" -F "file=@voice-transcription.zip" + +# Configure (per session, or '*' for all) +curl -X PUT http://localhost:2785/plugins/voice-transcription/config \ + -H "X-API-Key: $OPENWA_API_KEY" -H 'Content-Type: application/json' \ + -d '{"sttBaseUrl":"http://127.0.0.1:8000","model":"small","deliveryWebhookUrl":"http://127.0.0.1:5678/webhook/transcript"}' + +# Enable +curl -X POST http://localhost:2785/plugins/voice-transcription/enable \ + -H "X-API-Key: $OPENWA_API_KEY" +``` + +## Configuration + +| Key | Required | Default | Description | +| --- | -------- | ------- | ----------- | +| `sttBaseUrl` | yes | — | OpenAI-compatible STT base URL (`/v1/audio/transcriptions` is appended). Host must be in `net.allow`; localhost also needs `SSRF_ALLOWED_HOSTS`. | +| `sttApiKey` | no | — | Bearer key for hosted STT (Groq/OpenAI). Stored redacted. | +| `model` | no | `small` | Whisper model name. | +| `language` | no | _(auto)_ | BCP-47 hint; blank = auto-detect. | +| `provider` | no | `faster-whisper` | Label recorded in the delivered event. | +| `timeoutMs` | no | `20000` | STT request timeout (max 30000). | +| `enabledMessageTypes` | no | `["voice"]` | Add `audio` to also transcribe non-PTT audio (more cost). | +| `maxSizeBytes` | no | `16777216` | Skip audio larger than this (exact cost guard). | +| `maxPerHour` | no | `60` | Best-effort per-session hourly transcription cap. | +| `deliveryWebhookUrl` | cond. | — | Endpoint receiving the `message.transcription` event. Host must be in `net.allow`. Optional if you only use `chatDelivery`. | +| `deliverySecret` | no | — | Optional. HMAC-SHA256 signs the body in `X-OpenWA-Signature: sha256=` (same as core webhooks). Stored redacted. | +| `deliveryTimeoutMs` | no | `5000` | Delivery POST timeout. | +| `chatDelivery` | no | `off` | Also post the transcript into WhatsApp: `off` (webhook only) · `self` (note to your own number) · `reply` (quote-reply to the sender — visible to them). | + +## Compatibility + +- Engine-neutral: both Baileys and whatsapp-web.js materialize the audio before the hook fires, so the + plugin works on either. +- **Best-effort by design (no core changes).** Because a sandboxed plugin has no host-managed background + queue, transcription runs as an un-awaited task in the worker turn: it is **at-most-once while the worker + is alive**, has no backpressure, and the hourly/idempotency guards are best-effort (a truly simultaneous + engine re-fire can still double-call STT). For exactly-once, structured-event delivery, a future core + `message.transcription` event would be the upgrade path. + +## Security + +- **Outbound HTTP is allow-listed.** Calls go through the host's SSRF-guarded `ctx.net.fetch`; only hosts in + `net.allow` are reachable, and internal/loopback IPs stay blocked unless the operator opts in via + `SSRF_ALLOWED_HOSTS`. This plugin never ships `net.allow: ["*"]` — it carries your voice audio and an API + key, so egress is pinned to the STT and delivery hosts you configure. +- **Secrets** are stored redacted. `sttApiKey` is sent as a Bearer token only to the STT host; + `deliverySecret` is **never transmitted** — it HMAC-signs the body (`X-OpenWA-Signature`). +- **Treat the transcript as untrusted.** `transcription.text` is attacker-controlled speech — the event + marks it `untrusted: true`. A downstream LLM auto-responder MUST place it in a **user** role, never a + system/trusted context (a caller can *speak* injection instructions a typist never would). + +## Changelog + +See [CHANGELOG.md](./CHANGELOG.md). + +## License + +MIT diff --git a/voice-transcription/index.test.ts b/voice-transcription/index.test.ts new file mode 100644 index 0000000..167de1e --- /dev/null +++ b/voice-transcription/index.test.ts @@ -0,0 +1,91 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import type { PluginContext, HookContext, HookResult, IncomingMessage, PluginNetResponse } from '../types/openwa'; +import { VoiceTranscriptionPlugin } from './index.ts'; + +function voiceMsg(): IncomingMessage { + return { + id: 'm1', + from: 'x@s.whatsapp.net', + to: 'y@s.whatsapp.net', + chatId: 'c1@s.whatsapp.net', + body: '', + type: 'voice', + timestamp: 0, + fromMe: false, + isGroup: false, + media: { mimetype: 'audio/ogg; codecs=opus', data: Buffer.from('AUDIO').toString('base64'), sizeBytes: 5 }, + }; +} + +function makeStorage() { + const m = new Map(); + return { + get: async (k: string) => (m.has(k) ? m.get(k) : null), + set: async (k: string, v: unknown) => void m.set(k, v), + delete: async (k: string) => void m.delete(k), + list: async () => [...m.keys()], + }; +} + +function fakeContext(opts: { net: { fetch: (...a: unknown[]) => Promise }; config?: Record }) { + let hook: ((ctx: HookContext) => Promise) | undefined; + const ctx = { + pluginId: 'voice-transcription', + manifest: { id: 'voice-transcription' }, + config: { sttBaseUrl: 'http://stt', deliveryWebhookUrl: 'http://hook.local/in', ...opts.config }, + logger: { log() {}, debug() {}, warn() {}, error() {} }, + storage: makeStorage(), + registerHook: (event: string, handler: (c: HookContext) => Promise) => { + if (event === 'message:received') hook = handler; + }, + messages: {}, + engine: {}, + net: opts.net, + hookManager: {}, + } as unknown as PluginContext; + return { ctx, getHook: () => hook }; +} + +const engineCtx = (data: IncomingMessage): HookContext => ({ + event: 'message:received', + source: 'Engine', + sessionId: 's1', + timestamp: new Date(0), + data, +}); + +test('the message:received hook returns {continue:true} without awaiting STT (non-blocking)', async () => { + // ctx.net.fetch never resolves. If the hook awaited the STT round-trip, the call below would hang. + const net = { fetch: () => new Promise(() => {}) }; + const { ctx, getHook } = fakeContext({ net }); + const plugin = new VoiceTranscriptionPlugin(); + await plugin.onEnable(ctx); + const hook = getHook(); + assert.ok(hook, 'plugin must register a message:received hook'); + + const hookCall = hook(engineCtx(voiceMsg())); + const winner = await Promise.race([ + hookCall.then(r => ({ result: r })), + new Promise(res => setTimeout(() => res('timeout'), 250)), + ]); + assert.notEqual(winner, 'timeout', 'hook blocked on STT instead of returning immediately'); + assert.deepEqual((winner as { result: HookResult }).result, { continue: true }); +}); + +test('does not start transcription for non-Engine sources', async () => { + let fetched = false; + const net = { + fetch: async () => { + fetched = true; + return { ok: true, status: 200, statusText: '', headers: {}, body: '{"text":"x"}' } as PluginNetResponse; + }, + }; + const { ctx, getHook } = fakeContext({ net }); + const plugin = new VoiceTranscriptionPlugin(); + await plugin.onEnable(ctx); + const result = await getHook()!({ ...engineCtx(voiceMsg()), source: 'Dashboard' }); + assert.deepEqual(result, { continue: true }); + await new Promise(r => setImmediate(r)); // let any floated work run a tick + assert.equal(fetched, false); +}); diff --git a/voice-transcription/index.ts b/voice-transcription/index.ts new file mode 100644 index 0000000..0862058 --- /dev/null +++ b/voice-transcription/index.ts @@ -0,0 +1,120 @@ +/** + * Voice-note transcription extension. + * + * Registers on `message:received`, and for inbound voice notes runs speech-to-text OFF the dispatch + * critical path: the hook returns `{ continue: true }` synchronously and the STT call + delivery run as + * a deliberately un-awaited promise, so a slow transcription never blocks (or delays) message delivery. + * The transcript is delivered out-of-band as a `message.transcription` event POSTed to a configurable + * webhook — never echoed back into the contact's chat. Disabled until enabled via + * `POST /plugins/voice-transcription/enable`. + */ +import type { PluginContext, IPlugin, HookContext, HookResult, IncomingMessage } from '../types/openwa'; +import { OpenAiSttClient } from './openai-stt.client.ts'; +import { WebhookDelivery } from './webhook.delivery.ts'; +import { TranscriptionCoordinator, KvStore, ChatDeliveryMode } from './transcription.coordinator.ts'; + +function readString(cfg: Record, key: string, fallback: string): string { + const v = cfg[key]; + return typeof v === 'string' && v.length > 0 ? v : fallback; +} +function readOptionalString(cfg: Record, key: string): string | undefined { + const v = cfg[key]; + return typeof v === 'string' && v.length > 0 ? v : undefined; +} +function readNumber(cfg: Record, key: string, fallback: number): number { + const v = cfg[key]; + return typeof v === 'number' && Number.isFinite(v) ? v : fallback; +} +function readStringArray(cfg: Record, key: string, fallback: string[]): string[] { + const v = cfg[key]; + return Array.isArray(v) && v.every(x => typeof x === 'string') && v.length > 0 ? (v as string[]) : fallback; +} +function readChatDelivery(cfg: Record): ChatDeliveryMode { + const v = cfg['chatDelivery']; + return v === 'self' || v === 'reply' ? v : 'off'; +} + +export class VoiceTranscriptionPlugin implements IPlugin { + private coordinator: TranscriptionCoordinator | null = null; + + onEnable(context: PluginContext): Promise { + this.coordinator = this.build(context); + context.registerHook('message:received', ctx => + Promise.resolve(this.onMessage(ctx as HookContext)), + ); + if (!readOptionalString(context.config, 'deliveryWebhookUrl') && readChatDelivery(context.config) === 'off') { + context.logger.warn( + 'voice-transcription: no delivery configured — set deliveryWebhookUrl or chatDelivery, else transcripts have nowhere to go', + { action: 'transcription_no_delivery' }, + ); + } + context.logger.log('Voice transcription plugin enabled', { action: 'transcription_enabled' }); + return Promise.resolve(); + } + + onConfigChange(context: PluginContext): Promise { + // Rebuild so an edited config (new STT URL/key, delivery URL) applies without a disable/enable cycle. + this.coordinator = this.build(context); + context.logger.log('Voice transcription config updated', { action: 'transcription_config_changed' }); + return Promise.resolve(); + } + + onDisable(context: PluginContext): Promise { + this.coordinator = null; + context.logger.log('Voice transcription plugin disabled', { action: 'transcription_disabled' }); + return Promise.resolve(); + } + + private build(context: PluginContext): TranscriptionCoordinator { + const cfg = context.config; + const provider = new OpenAiSttClient({ + baseUrl: readString(cfg, 'sttBaseUrl', ''), + apiKey: readOptionalString(cfg, 'sttApiKey'), + model: readString(cfg, 'model', 'small'), + language: readOptionalString(cfg, 'language'), + timeoutMs: readNumber(cfg, 'timeoutMs', 20000), + net: context.net, + }); + const deliveryUrl = readString(cfg, 'deliveryWebhookUrl', ''); + const delivery = deliveryUrl + ? new WebhookDelivery({ + url: deliveryUrl, + secret: readOptionalString(cfg, 'deliverySecret'), + timeoutMs: readNumber(cfg, 'deliveryTimeoutMs', 5000), + net: context.net, + }) + : undefined; + const store: KvStore = { + get: key => context.storage.get(key), + set: (key, value) => context.storage.set(key, value), + }; + return new TranscriptionCoordinator({ + provider, + delivery, + chat: context.messages, // ChatSink — only used when chatDelivery !== 'off' + chatDelivery: readChatDelivery(cfg), + store, + config: { + enabledMessageTypes: readStringArray(cfg, 'enabledMessageTypes', ['voice']), + maxSizeBytes: readNumber(cfg, 'maxSizeBytes', 16 * 1024 * 1024), + maxPerHour: readNumber(cfg, 'maxPerHour', 60), + }, + providerLabel: readString(cfg, 'provider', 'faster-whisper'), + model: readString(cfg, 'model', 'small'), + logger: { warn: (m, meta) => context.logger.warn(m, meta) }, + }); + } + + /** + * Synchronous hook body: return `{ continue: true }` immediately and run transcription off the + * critical path. The coordinator is fail-open, so the floated promise needs no rejection handling. + */ + private onMessage(ctx: HookContext): HookResult { + if (this.coordinator && ctx.source === 'Engine' && ctx.sessionId) { + void this.coordinator.handle(ctx.sessionId, ctx.data); + } + return { continue: true }; + } +} + +export default VoiceTranscriptionPlugin; diff --git a/voice-transcription/manifest.json b/voice-transcription/manifest.json new file mode 100644 index 0000000..f4a4071 --- /dev/null +++ b/voice-transcription/manifest.json @@ -0,0 +1,72 @@ +{ + "id": "voice-transcription", + "name": "Voice Note Transcription", + "version": "1.0.0", + "type": "extension", + "main": "dist/index.js", + "description": "Transcribes inbound WhatsApp voice notes to text via an OpenAI-compatible speech-to-text backend (self-hosted Speaches/faster-whisper or hosted Groq/OpenAI) and delivers a `message.transcription` event to your webhook — so bots and AI can read and reply to audio. Off the message-delivery path; disabled until enabled.", + "author": "Yudhi Armyndharis ", + "license": "MIT", + "homepage": "https://github.com/rmyndharis/OpenWA-plugins/tree/main/voice-transcription", + "repository": "https://github.com/rmyndharis/OpenWA-plugins", + "keywords": ["transcription", "speech-to-text", "stt", "whisper", "voice", "audio", "whatsapp", "openwa"], + "status": "beta", + "minOpenWAVersion": "0.7.0", + "testedOpenWAVersion": "0.7.3", + "provides": ["transcription"], + "permissions": ["net:fetch"], + "net": { "allow": ["localhost", "127.0.0.1", "api.groq.com:443", "api.openai.com:443"] }, + "sessionScoped": true, + "sessions": ["*"], + "hooks": ["message:received"], + "configSchema": { + "type": "object", + "properties": { + "sttBaseUrl": { + "type": "string", + "title": "STT base URL", + "required": true, + "description": "Base URL of an OpenAI-compatible /v1/audio/transcriptions server — e.g. http://localhost:8000 (Speaches/faster-whisper) or https://api.groq.com/openai. Its host MUST be in this plugin's manifest net.allow; a localhost target ALSO requires SSRF_ALLOWED_HOSTS on the OpenWA host (use a literal 127.0.0.1)." + }, + "sttApiKey": { + "type": "string", + "title": "STT API key", + "secret": true, + "description": "Bearer key for a hosted backend (Groq/OpenAI). Leave blank for a local Speaches instance." + }, + "model": { "type": "string", "title": "Model", "default": "small", "description": "Whisper model name, e.g. small, base, whisper-large-v3-turbo." }, + "language": { "type": "string", "title": "Language hint", "default": "", "description": "Optional BCP-47 hint (e.g. es). Blank = auto-detect." }, + "provider": { "type": "string", "title": "Provider label", "default": "faster-whisper", "description": "Informational label recorded in the delivered event." }, + "timeoutMs": { "type": "number", "title": "STT timeout (ms)", "default": 20000, "min": 1000, "max": 30000, "description": "Per-request STT timeout. Runs off the message-delivery path, so it is bounded only by the host net.fetch ceiling (30000ms), not the 5s hook budget." }, + "enabledMessageTypes": { + "type": "array", + "title": "Message types to transcribe", + "default": ["voice"], + "items": { "type": "string" }, + "description": "Usually just voice (PTT). Add audio to also transcribe music/file audio (more cost)." + }, + "maxSizeBytes": { "type": "number", "title": "Max audio size (bytes)", "default": 16777216, "description": "Skip audio larger than this (exact cost guard)." }, + "maxPerHour": { "type": "number", "title": "Max transcriptions / hour / session", "default": 60, "description": "Best-effort per-session hourly cap to bound paid-API spend." }, + "deliveryWebhookUrl": { + "type": "string", + "title": "Delivery webhook URL", + "description": "Your endpoint that receives the message.transcription event (the bot/AI integration channel). Its host MUST be in this plugin's manifest net.allow. Optional if you only use in-chat delivery." + }, + "deliverySecret": { "type": "string", "title": "Delivery secret", "secret": true, "description": "Optional. When set, the body is HMAC-SHA256 signed in `X-OpenWA-Signature: sha256=` (same scheme as OpenWA core webhooks) so your endpoint can verify it." }, + "deliveryTimeoutMs": { "type": "number", "title": "Delivery timeout (ms)", "default": 5000, "min": 1000, "max": 30000 }, + "chatDelivery": { + "type": "string", + "title": "In-chat delivery", + "enum": ["off", "self", "reply"], + "default": "off", + "description": "Also post the transcript into WhatsApp: off (webhook only), self (a note to your own number), or reply (quote-reply to the sender — visible to them). Default off." + } + } + }, + "i18n": { + "es": { + "name": "Transcripción de Notas de Voz", + "description": "Transcribe las notas de voz entrantes de WhatsApp a texto mediante un backend de voz a texto compatible con OpenAI (Speaches/faster-whisper autoalojado o Groq/OpenAI) y entrega un evento message.transcription a tu webhook, para que los bots y la IA puedan leer y responder al audio. Fuera de la ruta de entrega de mensajes; desactivado hasta que se habilite." + } + } +} diff --git a/voice-transcription/multipart.test.ts b/voice-transcription/multipart.test.ts new file mode 100644 index 0000000..92a9f51 --- /dev/null +++ b/voice-transcription/multipart.test.ts @@ -0,0 +1,52 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { buildMultipartBody } from './multipart.ts'; + +test('assembles a text field followed by a binary file part, bytes intact', () => { + const boundary = 'X-BOUND-123'; + // Includes 0x00, 0x80 and 0xff — the bytes a UTF-8 string body would corrupt. + const audio = Uint8Array.from([0x00, 0x80, 0xff, 0x4f, 0x67, 0x67]); + + const body = buildMultipartBody( + boundary, + [{ name: 'model', value: 'small' }], + [{ name: 'file', filename: 'voice.ogg', contentType: 'audio/ogg', data: audio }], + ); + + const expected = Buffer.concat([ + Buffer.from(`--${boundary}\r\nContent-Disposition: form-data; name="model"\r\n\r\nsmall\r\n`), + Buffer.from( + `--${boundary}\r\nContent-Disposition: form-data; name="file"; filename="voice.ogg"\r\n` + + `Content-Type: audio/ogg\r\n\r\n`, + ), + Buffer.from(audio), + Buffer.from(`\r\n--${boundary}--\r\n`), + ]); + + assert.deepEqual(body, expected); +}); + +test('preserves high bytes (no UTF-8 expansion) — the load-bearing property', () => { + const audio = Uint8Array.from([0x80, 0xff, 0xfe]); // each would become 2 bytes if UTF-8 encoded + const body = buildMultipartBody( + 'b', + [], + [{ name: 'file', filename: 'v.ogg', contentType: 'audio/ogg', data: audio }], + ); + // The three raw bytes appear contiguously and unexpanded. + assert.ok(body.includes(Buffer.from(audio))); +}); + +test('a file-only body (no fields) is well-formed', () => { + const body = buildMultipartBody( + 'b', + [], + [{ name: 'file', filename: 'v.ogg', contentType: 'audio/ogg', data: Uint8Array.from([1, 2]) }], + ); + const expected = Buffer.concat([ + Buffer.from('--b\r\nContent-Disposition: form-data; name="file"; filename="v.ogg"\r\nContent-Type: audio/ogg\r\n\r\n'), + Buffer.from(Uint8Array.from([1, 2])), + Buffer.from('\r\n--b--\r\n'), + ]); + assert.deepEqual(body, expected); +}); diff --git a/voice-transcription/multipart.ts b/voice-transcription/multipart.ts new file mode 100644 index 0000000..fe86457 --- /dev/null +++ b/voice-transcription/multipart.ts @@ -0,0 +1,40 @@ +export interface MultipartField { + name: string; + value: string; +} + +export interface MultipartFilePart { + name: string; + filename: string; + contentType: string; + data: Uint8Array; +} + +/** + * Assemble a multipart/form-data request body as a Buffer. Binary file parts are concatenated as raw + * bytes (never string-encoded), so audio survives intact across the sandbox→host fetch boundary. + */ +export function buildMultipartBody( + boundary: string, + fields: MultipartField[], + files: MultipartFilePart[], +): Buffer { + const parts: Buffer[] = []; + for (const field of fields) { + parts.push( + Buffer.from(`--${boundary}\r\nContent-Disposition: form-data; name="${field.name}"\r\n\r\n${field.value}\r\n`), + ); + } + for (const file of files) { + parts.push( + Buffer.from( + `--${boundary}\r\nContent-Disposition: form-data; name="${file.name}"; filename="${file.filename}"\r\n` + + `Content-Type: ${file.contentType}\r\n\r\n`, + ), + ); + parts.push(Buffer.from(file.data)); // raw bytes — never string-encoded + parts.push(Buffer.from('\r\n')); + } + parts.push(Buffer.from(`--${boundary}--\r\n`)); + return Buffer.concat(parts); +} diff --git a/voice-transcription/openai-stt.client.test.ts b/voice-transcription/openai-stt.client.test.ts new file mode 100644 index 0000000..9828b9a --- /dev/null +++ b/voice-transcription/openai-stt.client.test.ts @@ -0,0 +1,142 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import type { PluginNetCapability, PluginNetRequestInit, PluginNetResponse } from '../types/openwa'; +import { OpenAiSttClient } from './openai-stt.client.ts'; + +function res(partial: { ok?: boolean; status?: number; body?: string }): PluginNetResponse { + return { + ok: partial.ok ?? true, + status: partial.status ?? 200, + statusText: '', + headers: {}, + body: partial.body ?? '{}', + text: async () => partial.body ?? '{}', + json: (async () => JSON.parse(partial.body ?? '{}')) as PluginNetResponse['json'], + arrayBuffer: async () => new ArrayBuffer(0), + }; +} + +function fakeNet(response: PluginNetResponse) { + const calls: Array<{ url: string; init: PluginNetRequestInit }> = []; + const net: PluginNetCapability = { + fetch: async (url: string, init?: PluginNetRequestInit) => { + calls.push({ url, init: init ?? {} }); + return response; + }, + }; + return { net, calls }; +} + +const body = (init: PluginNetRequestInit) => init.body as Buffer; + +test('posts to /v1/audio/transcriptions and returns the transcribed text', async () => { + const { net, calls } = fakeNet(res({ body: JSON.stringify({ text: 'hello there' }) })); + const c = new OpenAiSttClient({ baseUrl: 'http://stt:8000/', model: 'small', timeoutMs: 20000, net }); + const out = await c.transcribe(Uint8Array.from([1, 2, 3]), 'audio/ogg; codecs=opus'); + assert.equal(out.text, 'hello there'); + assert.equal(calls[0].url, 'http://stt:8000/v1/audio/transcriptions'); // trailing slash trimmed + assert.equal(calls[0].init.method, 'POST'); +}); + +test('uploads audio as a binary multipart Buffer with model + voice.ogg part, codecs stripped', async () => { + const { net, calls } = fakeNet(res({ body: '{"text":"x"}' })); + const c = new OpenAiSttClient({ baseUrl: 'http://stt:8000', model: 'base', timeoutMs: 1000, net }); + await c.transcribe(Uint8Array.from([0x80, 0xff, 0x4f]), 'audio/ogg; codecs=opus'); + const b = body(calls[0].init); + assert.ok(Buffer.isBuffer(b), 'body must be a binary Buffer, not a string'); + assert.ok(b.includes(Buffer.from(Uint8Array.from([0x80, 0xff, 0x4f]))), 'raw audio bytes intact'); + const text = b.toString('latin1'); + assert.ok(text.includes('name="model"') && text.includes('base')); + assert.ok(text.includes('filename="voice.ogg"')); + assert.ok(text.includes('Content-Type: audio/ogg') && !text.includes('codecs=opus')); + assert.match(calls[0].init.headers!['content-type'], /^multipart\/form-data; boundary=/); +}); + +test('includes the language field only when configured', async () => { + const withLang = fakeNet(res({ body: '{"text":"x"}' })); + await new OpenAiSttClient({ baseUrl: 'http://stt', model: 's', language: 'es', timeoutMs: 1000, net: withLang.net }) + .transcribe(Uint8Array.from([1]), 'audio/ogg'); + assert.ok(body(withLang.calls[0].init).toString('latin1').includes('name="language"')); + + const noLang = fakeNet(res({ body: '{"text":"x"}' })); + await new OpenAiSttClient({ baseUrl: 'http://stt', model: 's', timeoutMs: 1000, net: noLang.net }) + .transcribe(Uint8Array.from([1]), 'audio/ogg'); + assert.ok(!body(noLang.calls[0].init).toString('latin1').includes('name="language"')); +}); + +test('sets Authorization Bearer when an apiKey is configured, omits it otherwise', async () => { + const withKey = fakeNet(res({ body: '{"text":"x"}' })); + await new OpenAiSttClient({ baseUrl: 'http://stt', model: 's', apiKey: 'sk-1', timeoutMs: 1000, net: withKey.net }) + .transcribe(Uint8Array.from([1]), 'audio/ogg'); + assert.equal(withKey.calls[0].init.headers!['authorization'], 'Bearer sk-1'); + + const noKey = fakeNet(res({ body: '{"text":"x"}' })); + await new OpenAiSttClient({ baseUrl: 'http://stt', model: 's', timeoutMs: 1000, net: noKey.net }) + .transcribe(Uint8Array.from([1]), 'audio/ogg'); + assert.equal(noKey.calls[0].init.headers!['authorization'], undefined); +}); + +test('throws on a non-ok HTTP status', async () => { + const { net } = fakeNet(res({ ok: false, status: 500, body: 'boom' })); + const c = new OpenAiSttClient({ baseUrl: 'http://stt', model: 's', timeoutMs: 1000, net }); + await assert.rejects(c.transcribe(Uint8Array.from([1]), 'audio/ogg'), /500/); +}); + +test('throws when the response body has no text string', async () => { + const { net } = fakeNet(res({ body: '{"foo":1}' })); + const c = new OpenAiSttClient({ baseUrl: 'http://stt', model: 's', timeoutMs: 1000, net }); + await assert.rejects(c.transcribe(Uint8Array.from([1]), 'audio/ogg'), /text/); +}); + +function throwingNet() { + let n = 0; + const net: PluginNetCapability = { + fetch: async () => { + n++; + throw new Error('econnrefused'); + }, + }; + return { net, calls: () => n }; +} + +const a = Uint8Array.from([1]); + +test('opens the circuit after failureThreshold failures and short-circuits without hitting the network', async () => { + const { net, calls } = throwingNet(); + const c = new OpenAiSttClient({ baseUrl: 'http://stt', model: 's', timeoutMs: 1000, net, failureThreshold: 2 }); + await assert.rejects(c.transcribe(a, 'audio/ogg'), /econnrefused/); + await assert.rejects(c.transcribe(a, 'audio/ogg'), /econnrefused/); + await assert.rejects(c.transcribe(a, 'audio/ogg'), /circuit open/); // open → no network + assert.equal(calls(), 2); +}); + +test('a success resets the consecutive-failure counter (circuit stays closed)', async () => { + let mode: 'fail' | 'ok' = 'fail'; + const net: PluginNetCapability = { + fetch: async () => { + if (mode === 'fail') throw new Error('boom'); + return res({ body: '{"text":"x"}' }); + }, + }; + const c = new OpenAiSttClient({ baseUrl: 'http://stt', model: 's', timeoutMs: 1000, net, failureThreshold: 2 }); + await assert.rejects(c.transcribe(a, 'audio/ogg')); // failure 1 + mode = 'ok'; + await c.transcribe(a, 'audio/ogg'); // success → reset + mode = 'fail'; + await assert.rejects(c.transcribe(a, 'audio/ogg')); // failure 1 again, below threshold + assert.equal(c.isHealthy(), true); +}); + +test('the circuit re-closes after the cooldown elapses', async () => { + let t = 1000; + const { net, calls } = throwingNet(); + const c = new OpenAiSttClient({ + baseUrl: 'http://stt', model: 's', timeoutMs: 1000, net, failureThreshold: 1, cooldownMs: 5000, now: () => t, + }); + await assert.rejects(c.transcribe(a, 'audio/ogg'), /econnrefused/); // opens (threshold 1) + await assert.rejects(c.transcribe(a, 'audio/ogg'), /circuit open/); // open + assert.equal(calls(), 1); + t = 6001; // past cooldown + await assert.rejects(c.transcribe(a, 'audio/ogg'), /econnrefused/); // closed → hits net again + assert.equal(calls(), 2); +}); diff --git a/voice-transcription/openai-stt.client.ts b/voice-transcription/openai-stt.client.ts new file mode 100644 index 0000000..97085ef --- /dev/null +++ b/voice-transcription/openai-stt.client.ts @@ -0,0 +1,106 @@ +import type { PluginNetCapability } from '../types/openwa'; +import { buildMultipartBody, MultipartField } from './multipart.ts'; + +export interface SttResult { + text: string; + language?: string; +} + +export interface SttProvider { + transcribe(audio: Uint8Array, mimetype: string): Promise; +} + +export interface OpenAiSttOptions { + /** Base URL of an OpenAI-compatible STT server (e.g. http://localhost:8000 — Speaches/faster-whisper). */ + baseUrl: string; + apiKey?: string; + model: string; + /** Optional language hint (BCP-47); empty/undefined = auto-detect. */ + language?: string; + timeoutMs: number; + net: PluginNetCapability; + /** Consecutive failures before the circuit opens (default 5). */ + failureThreshold?: number; + /** How long the circuit stays open after tripping (default 30000ms). */ + cooldownMs?: number; + /** Injectable clock (defaults to Date.now). */ + now?: () => number; +} + +/** + * Calls an OpenAI-compatible `/v1/audio/transcriptions` endpoint over the host-proxied, SSRF-guarded + * `ctx.net.fetch`. The audio is uploaded as a binary multipart body (a Buffer) — it crosses the + * sandbox→host boundary via structuredClone intact, which a string body could not. + */ +export class OpenAiSttClient implements SttProvider { + private readonly base: string; + private readonly failureThreshold: number; + private readonly cooldownMs: number; + private readonly now: () => number; + private consecutiveFailures = 0; + private openUntil = 0; + + constructor(private readonly opts: OpenAiSttOptions) { + this.base = opts.baseUrl.replace(/\/+$/, ''); + this.failureThreshold = opts.failureThreshold ?? 5; + this.cooldownMs = opts.cooldownMs ?? 30000; + this.now = opts.now ?? (() => Date.now()); + } + + isHealthy(): boolean { + return this.consecutiveFailures < this.failureThreshold; + } + + async transcribe(audio: Uint8Array, mimetype: string): Promise { + // Circuit breaker: while open, fail fast without touching a known-bad backend. + if (this.now() < this.openUntil) { + throw new Error('STT circuit open'); + } + try { + const result = await this.doTranscribe(audio, mimetype); + this.consecutiveFailures = 0; + return result; + } catch (err) { + this.consecutiveFailures++; + if (this.consecutiveFailures >= this.failureThreshold) { + this.openUntil = this.now() + this.cooldownMs; + } + throw err; + } + } + + private async doTranscribe(audio: Uint8Array, mimetype: string): Promise { + const boundary = `----openwaFormBoundary${Math.random().toString(36).slice(2)}${Math.random().toString(36).slice(2)}`; + const fields: MultipartField[] = [ + { name: 'model', value: this.opts.model }, + { name: 'response_format', value: 'json' }, + ]; + if (this.opts.language) fields.push({ name: 'language', value: this.opts.language }); + + // Strip the codec suffix ("audio/ogg; codecs=opus" → "audio/ogg") and always name the part + // `voice.ogg`: OpenAI-compatible servers key the decoder off the filename extension and reject a + // bare ".opus", but accept ogg/oga. + const contentType = mimetype.split(';')[0].trim() || 'audio/ogg'; + const formBody = buildMultipartBody(boundary, fields, [ + { name: 'file', filename: 'voice.ogg', contentType, data: audio }, + ]); + + const headers: Record = { 'content-type': `multipart/form-data; boundary=${boundary}` }; + if (this.opts.apiKey) headers['authorization'] = `Bearer ${this.opts.apiKey}`; + + const response = await this.opts.net.fetch(`${this.base}/v1/audio/transcriptions`, { + method: 'POST', + headers, + body: formBody, + timeoutMs: this.opts.timeoutMs, + }); + if (!response.ok) { + throw new Error(`STT request failed: HTTP ${response.status}`); + } + const data = JSON.parse(response.body) as { text?: unknown; language?: unknown }; + if (typeof data?.text !== 'string') { + throw new Error('STT response contained no text'); + } + return { text: data.text, language: typeof data.language === 'string' ? data.language : undefined }; + } +} diff --git a/voice-transcription/transcription.coordinator.test.ts b/voice-transcription/transcription.coordinator.test.ts new file mode 100644 index 0000000..d77229c --- /dev/null +++ b/voice-transcription/transcription.coordinator.test.ts @@ -0,0 +1,207 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import type { IncomingMessage } from '../types/openwa'; +import type { SttResult, SttProvider } from './openai-stt.client.ts'; +import type { TranscriptionPayload } from './webhook.delivery.ts'; +import { + TranscriptionCoordinator, + KvStore, + TranscriptionConfig, + ChatSink, + ChatDeliveryMode, +} from './transcription.coordinator.ts'; + +function makeStore(): KvStore { + const m = new Map(); + return { + get: async (k: string) => (m.has(k) ? (m.get(k) as T) : null), + set: async (k, v) => void m.set(k, v), + }; +} + +function voiceMsg(over: Partial = {}): IncomingMessage { + return { + id: 'm1', + from: 'x@s.whatsapp.net', + to: 'y@s.whatsapp.net', + chatId: 'c1@s.whatsapp.net', + body: '', + type: 'voice', + timestamp: 0, + fromMe: false, + isGroup: false, + media: { mimetype: 'audio/ogg; codecs=opus', data: Buffer.from('AUDIO').toString('base64'), sizeBytes: 5 }, + ...over, + }; +} + +interface FakeProvider extends SttProvider { + calls: Array<{ audio: Uint8Array; mimetype: string }>; +} +type ChatSend = + | { kind: 'sendText'; sessionId: string; chatId: string; text: string } + | { kind: 'reply'; sessionId: string; chatId: string; quoted: string; text: string }; + +function setup(opts: { + result?: SttResult; + providerThrows?: boolean; + deliveryThrows?: boolean; + noDelivery?: boolean; + chatDelivery?: ChatDeliveryMode; + config?: Partial; + store?: KvStore; + now?: () => number; +}) { + const provider: FakeProvider = { + calls: [], + async transcribe(audio, mimetype) { + this.calls.push({ audio, mimetype }); + if (opts.providerThrows) throw new Error('stt down'); + return opts.result ?? { text: 'hola', language: 'es' }; + }, + }; + const deliveries: TranscriptionPayload[] = []; + const delivery = opts.noDelivery + ? undefined + : { + async deliver(e: TranscriptionPayload) { + deliveries.push(e); + if (opts.deliveryThrows) throw new Error('hook 502'); + }, + }; + const chatSends: ChatSend[] = []; + const chat: ChatSink = { + sendText: async (sessionId, chatId, text) => void chatSends.push({ kind: 'sendText', sessionId, chatId, text }), + reply: async (sessionId, chatId, quoted, text) => + void chatSends.push({ kind: 'reply', sessionId, chatId, quoted, text }), + }; + const warns: string[] = []; + const co = new TranscriptionCoordinator({ + provider, + delivery, + chat, + chatDelivery: opts.chatDelivery ?? 'off', + store: opts.store ?? makeStore(), + config: { enabledMessageTypes: ['voice'], maxSizeBytes: 1000, maxPerHour: 100, ...opts.config }, + providerLabel: 'faster-whisper', + model: 'small', + logger: { warn: m => void warns.push(m) }, + now: opts.now, + }); + return { co, provider, deliveries, chatSends, warns }; +} + +test('transcribes a voice note and delivers a completed event with decoded audio + payload', async () => { + const { co, provider, deliveries, chatSends } = setup({}); + await co.handle('s1', voiceMsg()); + assert.equal(provider.calls.length, 1); + assert.deepEqual([...provider.calls[0].audio], [...Buffer.from('AUDIO')]); + assert.equal(provider.calls[0].mimetype, 'audio/ogg; codecs=opus'); + assert.deepEqual(deliveries[0], { + event: 'message.transcription', + sessionId: 's1', + messageId: 'm1', + chatId: 'c1@s.whatsapp.net', + status: 'completed', + source: 'speech-to-text', + untrusted: true, + transcription: { text: 'hola', language: 'es', provider: 'faster-whisper', model: 'small' }, + }); + assert.equal(chatSends.length, 0); // chatDelivery off by default +}); + +test('skips a non-enabled message type — no STT, no event', async () => { + const { co, provider, deliveries } = setup({}); + await co.handle('s1', voiceMsg({ type: 'image' })); + assert.equal(provider.calls.length, 0); + assert.equal(deliveries.length, 0); +}); + +test('skips when media is absent — no event', async () => { + const { co, provider, deliveries } = setup({}); + await co.handle('s1', voiceMsg({ media: undefined })); + assert.equal(provider.calls.length, 0); + assert.equal(deliveries.length, 0); +}); + +test('emits a skipped event when media was omitted over the inbound cap', async () => { + const { co, provider, deliveries } = setup({}); + await co.handle('s1', voiceMsg({ media: { mimetype: 'audio/ogg', omitted: true, sizeBytes: 99999999 } })); + assert.equal(provider.calls.length, 0); + assert.equal(deliveries[0].status, 'skipped'); + assert.equal(deliveries[0].reason, 'media_unavailable'); +}); + +test('emits a skipped event when the audio exceeds maxSizeBytes', async () => { + const { co, provider, deliveries } = setup({ config: { maxSizeBytes: 3 } }); // 'AUDIO' decodes to 5 bytes + await co.handle('s1', voiceMsg()); + assert.equal(provider.calls.length, 0); + assert.equal(deliveries[0].status, 'skipped'); + assert.equal(deliveries[0].reason, 'too_large'); +}); + +test('idempotency: the same messageId is processed once (one STT call, one event)', async () => { + const { co, provider, deliveries } = setup({}); + await co.handle('s1', voiceMsg()); + await co.handle('s1', voiceMsg()); // re-fire + assert.equal(provider.calls.length, 1); + assert.equal(deliveries.length, 1); +}); + +test('rate limit: emits a skipped event once maxPerHour is reached, with no more STT', async () => { + const { co, provider, deliveries } = setup({ config: { maxPerHour: 2 }, now: () => 0 }); + for (const id of ['a', 'b', 'c']) await co.handle('s1', voiceMsg({ id })); + assert.equal(provider.calls.length, 2); + assert.equal(deliveries[deliveries.length - 1].status, 'skipped'); + assert.equal(deliveries[deliveries.length - 1].reason, 'rate_limited'); +}); + +test('emits a skipped event when the transcript is empty/whitespace', async () => { + const { co, deliveries } = setup({ result: { text: ' ' } }); + await co.handle('s1', voiceMsg()); + assert.equal(deliveries[0].status, 'skipped'); + assert.equal(deliveries[0].reason, 'empty'); +}); + +test('emits a failed event (and never throws) when STT errors', async () => { + const { co, deliveries, warns } = setup({ providerThrows: true }); + await assert.doesNotReject(co.handle('s1', voiceMsg())); + assert.equal(deliveries[0].status, 'failed'); + assert.ok(warns.length >= 1); +}); + +test('fail-open: a delivery error is swallowed and warned', async () => { + const { co, warns } = setup({ deliveryThrows: true }); + await assert.doesNotReject(co.handle('s1', voiceMsg())); + assert.ok(warns.length >= 1); +}); + +test('chatDelivery=self sends the transcript to the bot own number (msg.to)', async () => { + const { co, chatSends } = setup({ chatDelivery: 'self' }); + await co.handle('s1', voiceMsg()); + assert.deepEqual(chatSends[0], { kind: 'sendText', sessionId: 's1', chatId: 'y@s.whatsapp.net', text: 'hola' }); +}); + +test('chatDelivery=reply quote-replies in the original chat', async () => { + const { co, chatSends } = setup({ chatDelivery: 'reply' }); + await co.handle('s1', voiceMsg()); + assert.deepEqual(chatSends[0], { + kind: 'reply', + sessionId: 's1', + chatId: 'c1@s.whatsapp.net', + quoted: 'm1', + text: 'hola', + }); +}); + +test('chat delivery does not fire for a skipped/failed outcome', async () => { + const { co, chatSends } = setup({ chatDelivery: 'self', config: { maxSizeBytes: 3 } }); // oversize → skipped + await co.handle('s1', voiceMsg()); + assert.equal(chatSends.length, 0); +}); + +test('works chat-only when no webhook delivery is configured', async () => { + const { co, chatSends } = setup({ noDelivery: true, chatDelivery: 'self' }); + await assert.doesNotReject(co.handle('s1', voiceMsg())); + assert.equal(chatSends.length, 1); +}); diff --git a/voice-transcription/transcription.coordinator.ts b/voice-transcription/transcription.coordinator.ts new file mode 100644 index 0000000..baa7bf4 --- /dev/null +++ b/voice-transcription/transcription.coordinator.ts @@ -0,0 +1,158 @@ +import type { IncomingMessage } from '../types/openwa'; +import type { SttProvider, SttResult } from './openai-stt.client.ts'; +import type { TranscriptDelivery, TranscriptionPayload } from './webhook.delivery.ts'; + +/** Minimal KV surface the coordinator needs (adapted from `ctx.storage` by the plugin). */ +export interface KvStore { + get(key: string): Promise; + set(key: string, value: unknown): Promise; +} + +export interface CoordinatorLogger { + warn(message: string, meta?: Record): void; +} + +/** Send the transcript into a WhatsApp chat (adapted from `ctx.messages`). */ +export interface ChatSink { + sendText(sessionId: string, chatId: string, text: string): Promise; + reply(sessionId: string, chatId: string, quotedMessageId: string, text: string): Promise; +} + +/** off = no chat message; self = note to the bot's own number; reply = quote-reply to the sender. */ +export type ChatDeliveryMode = 'off' | 'self' | 'reply'; + +export interface TranscriptionConfig { + /** Message types to transcribe, e.g. ['voice'] (PTT). */ + enabledMessageTypes: string[]; + /** Skip audio larger than this (decoded bytes) — the exact cost guard. */ + maxSizeBytes: number; + /** Best-effort per-session hourly cap on transcriptions. */ + maxPerHour: number; +} + +export interface CoordinatorDeps { + provider: SttProvider; + /** Webhook sink. Optional: omit for chat-only operation. */ + delivery?: TranscriptDelivery; + /** Chat sink (used when chatDelivery !== 'off'). */ + chat?: ChatSink; + chatDelivery: ChatDeliveryMode; + store: KvStore; + config: TranscriptionConfig; + /** Label recorded in the delivered event, e.g. 'faster-whisper'. */ + providerLabel: string; + model: string; + logger: CoordinatorLogger; + /** Injectable clock for the hourly rate-limit bucket (defaults to Date.now). */ + now?: () => number; +} + +/** + * The framework-agnostic core: decide whether to transcribe an inbound message, run the guards (type, + * size, idempotency, rate limit), call STT, and report the outcome — a `completed` event with the + * transcript, or a `failed`/`skipped` event explaining why. Fail-open throughout: never throws to the + * caller, so a transcription failure can never disrupt message delivery. + */ +export class TranscriptionCoordinator { + private readonly now: () => number; + + constructor(private readonly deps: CoordinatorDeps) { + this.now = deps.now ?? (() => Date.now()); + } + + async handle(sessionId: string, msg: IncomingMessage): Promise { + const { config, store, logger } = this.deps; + try { + // Not our concern → no event at all. + if (!config.enabledMessageTypes.includes(msg.type)) return; + if (!msg.media) return; + + // Idempotency first, so every outcome (including skips) fires at most once per message id. + // Best-effort: get-then-set is not atomic across a truly simultaneous #466 re-fire (documented). + const seenKey = `seen:${sessionId}:${msg.id}`; + if (await store.get(seenKey)) return; + await store.set(seenKey, 1); + + const media = msg.media; + if (media.omitted || !media.data) { + await this.emit(sessionId, msg, { status: 'skipped', reason: 'media_unavailable' }); + return; + } + if (!media.mimetype || !media.mimetype.startsWith('audio/')) return; // defensive: not audio + + const audio = Buffer.from(media.data, 'base64'); + if (audio.byteLength > config.maxSizeBytes) { + await this.emit(sessionId, msg, { status: 'skipped', reason: 'too_large' }); + return; + } + + const rateKey = `rate:${sessionId}:${Math.floor(this.now() / 3_600_000)}`; + const count = (await store.get(rateKey)) ?? 0; + if (count >= config.maxPerHour) { + await this.emit(sessionId, msg, { status: 'skipped', reason: 'rate_limited' }); + return; + } + await store.set(rateKey, count + 1); + + let result: SttResult; + try { + result = await this.deps.provider.transcribe(audio, media.mimetype); + } catch (err) { + await this.emit(sessionId, msg, { status: 'failed', reason: err instanceof Error ? err.message : String(err) }); + return; + } + if (!result.text.trim()) { + await this.emit(sessionId, msg, { status: 'skipped', reason: 'empty' }); + return; + } + + await this.emit(sessionId, msg, { + status: 'completed', + text: result.text, + transcription: { + text: result.text, + language: result.language, + provider: this.deps.providerLabel, + model: this.deps.model, + }, + }); + } catch (err) { + // Fail-open: a transcription/delivery failure must never disrupt message delivery. + logger.warn('Transcription failed (skipped)', { + messageId: msg.id, + error: err instanceof Error ? err.message : String(err), + }); + } + } + + private async emit( + sessionId: string, + msg: IncomingMessage, + o: { + status: TranscriptionPayload['status']; + reason?: string; + text?: string; + transcription?: NonNullable; + }, + ): Promise { + if (o.status !== 'completed') { + this.deps.logger.warn(`Transcription ${o.status}: ${o.reason}`, { messageId: msg.id }); + } + const payload: TranscriptionPayload = { + event: 'message.transcription', + sessionId, + messageId: msg.id, + chatId: msg.chatId, + status: o.status, + source: 'speech-to-text', + untrusted: true, + ...(o.reason ? { reason: o.reason } : {}), + ...(o.transcription ? { transcription: o.transcription } : {}), + }; + if (this.deps.delivery) await this.deps.delivery.deliver(payload); + if (o.status === 'completed' && o.text && this.deps.chat && this.deps.chatDelivery !== 'off') { + if (this.deps.chatDelivery === 'self') await this.deps.chat.sendText(sessionId, msg.to, o.text); + else await this.deps.chat.reply(sessionId, msg.chatId, msg.id, o.text); + } + } +} diff --git a/voice-transcription/webhook.delivery.test.ts b/voice-transcription/webhook.delivery.test.ts new file mode 100644 index 0000000..e7aad1f --- /dev/null +++ b/voice-transcription/webhook.delivery.test.ts @@ -0,0 +1,72 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createHmac } from 'node:crypto'; +import type { PluginNetCapability, PluginNetRequestInit, PluginNetResponse } from '../types/openwa'; +import { WebhookDelivery, TranscriptionPayload } from './webhook.delivery.ts'; + +function res(partial: { ok?: boolean; status?: number }): PluginNetResponse { + return { + ok: partial.ok ?? true, + status: partial.status ?? 200, + statusText: '', + headers: {}, + body: '', + text: async () => '', + json: (async () => ({})) as PluginNetResponse['json'], + arrayBuffer: async () => new ArrayBuffer(0), + }; +} + +function fakeNet(response: PluginNetResponse) { + const calls: Array<{ url: string; init: PluginNetRequestInit }> = []; + const net: PluginNetCapability = { + fetch: async (url: string, init?: PluginNetRequestInit) => { + calls.push({ url, init: init ?? {} }); + return response; + }, + }; + return { net, calls }; +} + +const payload: TranscriptionPayload = { + event: 'message.transcription', + sessionId: 's1', + messageId: 'm1', + chatId: 'c1@s.whatsapp.net', + status: 'completed', + source: 'speech-to-text', + untrusted: true, + transcription: { text: 'hola mundo', language: 'es', provider: 'faster-whisper', model: 'small' }, +}; + +test('POSTs the payload as JSON to the configured url', async () => { + const { net, calls } = fakeNet(res({})); + await new WebhookDelivery({ url: 'http://hook.local/in', timeoutMs: 5000, net }).deliver(payload); + assert.equal(calls[0].url, 'http://hook.local/in'); + assert.equal(calls[0].init.method, 'POST'); + assert.equal(calls[0].init.headers!['content-type'], 'application/json'); + assert.deepEqual(JSON.parse(calls[0].init.body as string), payload); +}); + +test('signs the body with HMAC-SHA256 in X-OpenWA-Signature when a secret is configured', async () => { + const { net, calls } = fakeNet(res({})); + await new WebhookDelivery({ url: 'http://hook.local/in', secret: 'shh', timeoutMs: 5000, net }).deliver(payload); + const sentBody = calls[0].init.body as string; + const expected = `sha256=${createHmac('sha256', 'shh').update(sentBody).digest('hex')}`; + assert.equal(calls[0].init.headers!['X-OpenWA-Signature'], expected); + assert.equal(calls[0].init.headers!['authorization'], undefined); // bearer replaced by the signature +}); + +test('omits the signature header when no secret is configured', async () => { + const { net, calls } = fakeNet(res({})); + await new WebhookDelivery({ url: 'http://hook.local/in', timeoutMs: 5000, net }).deliver(payload); + assert.equal(calls[0].init.headers!['X-OpenWA-Signature'], undefined); +}); + +test('throws on a non-ok status so the caller can log the miss', async () => { + const { net } = fakeNet(res({ ok: false, status: 502 })); + await assert.rejects( + new WebhookDelivery({ url: 'http://hook.local/in', timeoutMs: 5000, net }).deliver(payload), + /502/, + ); +}); diff --git a/voice-transcription/webhook.delivery.ts b/voice-transcription/webhook.delivery.ts new file mode 100644 index 0000000..3f9e038 --- /dev/null +++ b/voice-transcription/webhook.delivery.ts @@ -0,0 +1,54 @@ +import { createHmac } from 'node:crypto'; +import type { PluginNetCapability } from '../types/openwa'; + +/** The out-of-band event the plugin POSTs to an integrator's URL for an inbound voice note. */ +export interface TranscriptionPayload { + event: 'message.transcription'; + sessionId: string; + messageId: string; + chatId: string; + /** completed = transcript present; failed = STT errored; skipped = not transcribed (too large, rate-limited, empty). */ + status: 'completed' | 'failed' | 'skipped'; + source: 'speech-to-text'; + /** The transcript is attacker-controlled speech — downstream consumers MUST treat it as user input. */ + untrusted: true; + /** Why the note was skipped/failed (absent for completed). */ + reason?: string; + /** Present only when status is completed. */ + transcription?: { text: string; language?: string; provider: string; model: string }; +} + +export interface TranscriptDelivery { + deliver(payload: TranscriptionPayload): Promise; +} + +export interface WebhookDeliveryOptions { + url: string; + /** Optional shared secret. When set, the body is HMAC-SHA256 signed in `X-OpenWA-Signature` (same + * scheme as OpenWA's core webhooks: `sha256=`), so an existing receiver verifies it identically. */ + secret?: string; + timeoutMs: number; + net: PluginNetCapability; +} + +/** Delivers the transcription event as a JSON POST through the host-proxied, SSRF-guarded `ctx.net.fetch`. */ +export class WebhookDelivery implements TranscriptDelivery { + constructor(private readonly opts: WebhookDeliveryOptions) {} + + async deliver(payload: TranscriptionPayload): Promise { + const body = JSON.stringify(payload); + const headers: Record = { 'content-type': 'application/json' }; + if (this.opts.secret) { + headers['X-OpenWA-Signature'] = `sha256=${createHmac('sha256', this.opts.secret).update(body).digest('hex')}`; + } + const response = await this.opts.net.fetch(this.opts.url, { + method: 'POST', + headers, + body, + timeoutMs: this.opts.timeoutMs, + }); + if (!response.ok) { + throw new Error(`transcription delivery failed: HTTP ${response.status}`); + } + } +}