diff --git a/.env.docker.example b/.env.docker.example index 10a35ab..9b44a4e 100644 --- a/.env.docker.example +++ b/.env.docker.example @@ -13,6 +13,13 @@ ELEVENLABS_API_KEY= CALL_SCRIBE_DISCORD_GUILD_ID= CALL_SCRIBE_DISCORD_CHANNEL_ID= +# Hosted alternative to the static guild/channel values above. Set both or +# neither; store the workload token in a secrets manager in production. +# CALL_SCRIBE_HOSTED_CONTROL_PLANE_URL=https://callscribe.example.com +# CALL_SCRIBE_HOSTED_WORKLOAD_TOKEN= +# CALL_SCRIBE_HOSTED_OUTBOX_ENCRYPTION_KEY= +# CALL_SCRIBE_HOSTED_WORKER_ID=call-scribe-worker + # Optional overrides. # CALL_SCRIBE_STT_PROVIDER=eleven-labs # ELEVENLABS_STT_TIMEOUT_SECONDS=900 diff --git a/.env.example b/.env.example index 6e147ee..c1ff622 100644 --- a/.env.example +++ b/.env.example @@ -41,3 +41,14 @@ DISCORD_TOKEN= CALL_SCRIBE_DISCORD_USER_ID= CALL_SCRIBE_DISCORD_GUILD_ID= CALL_SCRIBE_DISCORD_CHANNEL_ID= + +# Hosted worker mode (optional). Set URL and token together. The workload token +# is a dedicated machine credential; do not reuse a Discord or user token. +# Hosted mode cannot be combined with static guild/channel settings, polls +# fail-closed policy, and starts only from an explicit durable command. +CALL_SCRIBE_HOSTED_CONTROL_PLANE_URL= +CALL_SCRIBE_HOSTED_WORKLOAD_TOKEN= +CALL_SCRIBE_HOSTED_OUTBOX_ENCRYPTION_KEY= +CALL_SCRIBE_HOSTED_WORKER_ID=call-scribe-worker +CALL_SCRIBE_HOSTED_POLL_SECONDS=15 +CALL_SCRIBE_HOSTED_MAX_STALENESS_SECONDS=60 diff --git a/Cargo.lock b/Cargo.lock index 65bbcb0..0b14a85 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -322,6 +322,7 @@ dependencies = [ "async-trait", "axum", "base64", + "chacha20poly1305", "chrono", "clap", "dashmap", diff --git a/Cargo.toml b/Cargo.toml index ca0580f..41611ee 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,7 +11,7 @@ categories = ["command-line-utilities", "multimedia::audio"] [features] default = [] -discord = ["dep:dashmap", "dep:hound", "dep:rustls", "dep:serenity", "dep:songbird"] +discord = ["dep:chacha20poly1305", "dep:dashmap", "dep:hound", "dep:rustls", "dep:serenity", "dep:songbird"] [dependencies] anyhow = "1.0" @@ -19,6 +19,7 @@ async-trait = "0.1" axum = { version = "0.8", features = ["macros"] } base64 = "0.22" chrono = { version = "0.4", features = ["serde"] } +chacha20poly1305 = { version = "0.10", optional = true } clap = { version = "4.5", features = ["derive", "env"] } dashmap = { version = "6.1", optional = true } hound = { version = "3.5", optional = true } diff --git a/README.md b/README.md index ae2adc6..ceed195 100644 --- a/README.md +++ b/README.md @@ -133,6 +133,12 @@ Phone calls need a telephony adapter such as Twilio recording webhooks. The webh Always get participant consent before recording or transcribing calls. +For a separate SaaS control plane, use the fail-closed hosted adapter described +in [docs/deployment/hosted-worker.md](docs/deployment/hosted-worker.md). It keeps +billing and product identity outside this public recorder, disables hosted +occupancy auto-start, and requires fresh entitlement/channel/consent/cap policy +plus an explicit durable start command. + ## Troubleshooting Capture Gaps Switching Discord from desktop to phone should still be recorded by the bot as long as the phone client is actually joined to the configured guild voice/stage channel and transmitting audio. The bot records what Discord receives in that channel; it cannot recover audio that only reached a local screen recorder, a handset, Bluetooth route, or a separate phone call. diff --git a/charts/call-scribe/templates/deployment.yaml b/charts/call-scribe/templates/deployment.yaml index 7cb6870..a63a671 100644 --- a/charts/call-scribe/templates/deployment.yaml +++ b/charts/call-scribe/templates/deployment.yaml @@ -66,6 +66,26 @@ spec: value: {{ .Values.config.captureMode | quote }} - name: CALL_SCRIBE_ORGANIZATION_ID value: {{ .Values.config.organizationId | quote }} + {{- if .Values.hosted.enabled }} + - name: CALL_SCRIBE_HOSTED_CONTROL_PLANE_URL + value: {{ required "hosted.controlPlaneUrl is required when hosted.enabled=true" .Values.hosted.controlPlaneUrl | quote }} + - name: CALL_SCRIBE_HOSTED_WORKER_ID + value: {{ .Values.hosted.workerId | quote }} + - name: CALL_SCRIBE_HOSTED_POLL_SECONDS + value: {{ .Values.hosted.pollSeconds | quote }} + - name: CALL_SCRIBE_HOSTED_MAX_STALENESS_SECONDS + value: {{ .Values.hosted.maxStalenessSeconds | quote }} + - name: CALL_SCRIBE_HOSTED_WORKLOAD_TOKEN + valueFrom: + secretKeyRef: + name: {{ required "hosted.workloadCredentials.existingSecret is required when hosted.enabled=true" .Values.hosted.workloadCredentials.existingSecret | quote }} + key: {{ .Values.hosted.workloadCredentials.tokenKey | quote }} + - name: CALL_SCRIBE_HOSTED_OUTBOX_ENCRYPTION_KEY + valueFrom: + secretKeyRef: + name: {{ .Values.hosted.workloadCredentials.existingSecret | quote }} + key: {{ .Values.hosted.workloadCredentials.outboxEncryptionKey | quote }} + {{- end }} - name: ELEVENLABS_STT_TIMEOUT_SECONDS value: {{ .Values.config.sttTimeoutSeconds | quote }} - name: ELEVENLABS_STT_DIARIZE diff --git a/charts/call-scribe/values.yaml b/charts/call-scribe/values.yaml index a8dba43..d397812 100644 --- a/charts/call-scribe/values.yaml +++ b/charts/call-scribe/values.yaml @@ -31,6 +31,20 @@ config: captureMode: record-only organizationId: org_private_alpha +# Optional SaaS adapter. The public worker contains no billing logic; it accepts +# only the control plane's short-lived recording policy and durable commands. +# Hosted mode cannot be combined with static Discord guild/channel settings. +hosted: + enabled: false + controlPlaneUrl: "" + workerId: call-scribe-worker + pollSeconds: 15 + maxStalenessSeconds: 60 + workloadCredentials: + existingSecret: "" + tokenKey: CALL_SCRIBE_HOSTED_WORKLOAD_TOKEN + outboxEncryptionKey: CALL_SCRIBE_HOSTED_OUTBOX_ENCRYPTION_KEY + api: enabled: false replicaCount: 1 diff --git a/docs/deployment/hosted-worker.md b/docs/deployment/hosted-worker.md new file mode 100644 index 0000000..a26a070 --- /dev/null +++ b/docs/deployment/hosted-worker.md @@ -0,0 +1,228 @@ +# Hosted worker adapter + +The public Call Scribe worker can be attached to a separate hosted control +plane without embedding billing, identity, or proprietary product logic in the +open-source recorder. Self-hosted operation remains the default. + +## Safety boundary + +Hosted mode is enabled only when both +`CALL_SCRIBE_HOSTED_CONTROL_PLANE_URL` and +`CALL_SCRIBE_HOSTED_WORKLOAD_TOKEN` are present, together with the independent +`CALL_SCRIBE_HOSTED_OUTBOX_ENCRYPTION_KEY`. The URL must use HTTPS except for +loopback development. Do not reuse a Discord bot token, browser session, +provider API key, or the workload credential as the outbox key. +Hosted mode also requires `CALL_SCRIBE_DATABASE_URL` for durable command +idempotency across worker restarts. + +The worker does not start hosted recordings from voice-channel occupancy. +Every start requires a durable `record_start` command. Immediately before +joining voice, the worker also requires a fresh guild policy with all of these +conditions: + +- the subscription entitlement is active; +- recording is explicitly enabled; +- the requested channel is approved; +- the supported `explicit_command` consent mode and durable notice evidence are configured; +- a notice channel is configured; +- a retention period from 1 through 365 days is configured; +- the monthly recording cap exists and has remaining seconds; and +- at least one human participant is currently in the requested channel. + +Missing configuration, a failed first fetch, stale configuration, malformed +identifiers, an exhausted cap, or a revoked entitlement denies capture. A +policy refresh that revokes an active session—or removes its guild from the +full replacement snapshot—makes the worker leave, clears the pending start +request, and finalizes the retained audio. Configuration is held in memory and +replaced as one snapshot; billing data is not stored by the worker. + +## Control-plane contract + +All requests use `Authorization: Bearer ` and include +`X-Call-Scribe-Worker-Id`. The worker rejects tokens shorter than 32 bytes; +the server must bind the token to the configured worker ID and should keep +these internal routes on private ingress or require workload mTLS. + +This incremental adapter does not yet implement a hosted storage destination. +Every current provider (`customer_s3`, `customer_r2`, `google_drive`, and +`managed_transient`) therefore fails closed at the worker, even when the +control plane reports it as ready. Hosted recording remains disabled until a +provider adapter uploads, verifies, and deletes or retains transient local data +according to policy. Self-hosted local/PVC capture is unchanged. + +`GET /internal/v1/worker/guild-configurations` + +```json +{ + "revision": "cfg_01", + "guilds": [ + { + "guildId": "123", + "organizationId": "org_01", + "entitlementActive": true, + "approvedChannelIds": ["456"], + "noticeChannelId": "789", + "consentMode": "explicit_command", + "consentPolicyVersion": "v1", + "consentNoticeTemplate": "This call is being recorded with advance notice.", + "retentionDays": 30, + "recordingEnabled": true, + "monthlyRecordingSecondsCap": 36000, + "remainingRecordingSeconds": 35900, + "storageProvider": "customer_s3", + "storageDestinationLabel": "Customer archive", + "ready": true, + "blockedReasons": [], + "desiredRecordingGeneration": 42 + } + ] +} +``` + +`POST /internal/v1/worker/commands/lease` with +`{"workerId":"call-scribe-worker","limit":10}` returns: + +```json +{ + "commands": [ + { + "id": "cmd_01", + "commandKind": "record_start", + "guildId": "123", + "channelId": "456", + "leaseToken": "opaque-command-lease", + "leaseExpiresAt": "2026-08-12T18:02:00Z", + "generation": 42, + "recordingNoticeId": "notice-uuid" + } + ] +} +``` + +Supported kinds are `record_start` and `record_stop`. The worker posts +the terminal result to +`POST /internal/v1/worker/commands//complete`: + +```json +{ + "leaseToken": "opaque-command-lease", + "success": true, + "result": {"recordingId": "recording-uuid"} +} +``` + +Failed completions use `success:false` and a bounded error-only result. +Completion results are limited to the fields `code`, `message`, `recordingId`, +and `durationSeconds` and to 1,024 encoded bytes. +The control plane owns command durability, redelivery until acknowledgement, +authorization of the initiating human, and posting the consent notice before +enqueueing a start command. The worker's policy checks are an additional +fail-closed enforcement layer, not evidence that notice was actually posted. +Every leased start must carry the durable `recordingNoticeId` validated by the +control plane; the worker rejects a start without it. +The control plane increments `desiredRecordingGeneration` whenever desired +recording state changes and includes that generation on commands. The worker +rejects starts that do not match the fresh policy generation, stops an older +active generation when policy advances, and never lets an older stop cancel a +newer capture. The server must cancel older queued/leased starts when accepting +a stop and lease at most one state-changing command per guild. + +Before joining voice, the worker reserves usage through +`POST /internal/v1/worker/usage/reservations` with +`{"commandId":"cmd_01","requestedSeconds":3600}`. The control plane derives +the guild, generation, channel, +and notice evidence from the leased command and atomically revalidates all +gates. It returns an opaque reservation ID, lease +token, bounded reserved seconds, and expiry. The worker stops no later than the +reserved duration. It also posts `{"leaseToken":"..."}` to +`/reservations//heartbeat` at least every ten seconds and requires an exact +`{"reservationId":"...","expiresAt":"..."}` response. A heartbeat timeout, +non-200 response, mismatched reservation, malformed timestamp, or expiry within +the 15-second safety margin synchronously closes the WAV writer and stops the +capture. Expiry beyond the 90-second contract (allowing five seconds for clock +skew) is also rejected. A valid response may extend or shorten authority; the +returned expiry replaces the worker's prior deadline and is persisted with +crash-recovery ownership before the heartbeat succeeds locally. An independent +one-second local watchdog enforces that stored deadline even if the shared +heartbeat monitor exits unexpectedly. + +A finalized recording is sent to `/reservations//consume` with its +recording ID and audio-derived actual duration; failed starts call +`/reservations//release`. The server must use its own trusted time for +billing periods, make recording IDs idempotent only for the same reservation, +reject invalid settlement, and serialize heartbeat, reserve, consume, and +release against one guild-level quota lock. +The worker requests at most 3,600 seconds per reservation even when the guild +has more monthly usage remaining. + +The server grants a rolling capture authority no longer than 90 seconds and +revalidates current entitlement, suspension, privacy, and canonical quota state +on every heartbeat. When authority ends, the server keeps the reservation in a +bounded stopping state for final settlement. The current contract allows 30 +minutes for that settlement; this grace permits only the exact final consume +and never authorizes more recording. While a matching command/generation holds +an active reservation, the worker may continue through only the snapshot's +`usage_cap_exhausted` blocker and relies on the heartbeat plus local reserved +duration fence. It still stops for stale configuration, entitlement or privacy +revocation, consent/channel/storage changes, any other blocker, or a newer +generation. + +Final usage is written before delivery to a durable outbox. The reservation +lease token is encrypted with ChaCha20-Poly1305 under the dedicated stable key, +with the reservation ID as associated data; plaintext lease tokens are not +stored. The worker retries idempotent consumption every 30 seconds only inside +`expiresAt`, then marks the entry terminal and emits an operator-reconciliation +alert instead of retrying forever. `expiresAt` from the heartbeat is the capture +deadline; the encrypted outbox derives its own deadline by adding only the +30-minute settlement grace. Keep the outbox key available through that bounded +settlement window when rotating it. + +Before joining Discord voice, the worker also persists an encrypted active +reservation recovery row containing the recording ID and mixed-audio base WAV +path. Each process uses a unique runtime owner ID. At least every ten seconds it +first renews control-plane authority, then atomically persists the returned +expiry with its database-visible ownership heartbeat. Another replica can claim +only an owner stale for at least 60 seconds, and every retry/delete is fenced by +the current claim token. During startup, Discord join and each database/handler +step are bounded; both control-plane authority and database ownership are +renewed once more before voice handlers are enabled. Either heartbeat failure +or an ownership mismatch makes the old process synchronously close its hosted +WAV writers before any Discord, network, or database cleanup, so a partitioned +replica cannot keep recording past valid authority. + +WAV headers are checkpointed every five seconds. After a process or pod crash, +the next worker derives usage only from valid, contiguous mixed-audio WAV +segment headers. It never derives billable duration from `startedAt` to the +current wall clock. The recovered duration is rounded up to a whole second, +clamped to both `reservedSeconds` and the last persisted capture-authority +window, and idempotently moved to the normal usage outbox; a zero-duration +capture releases the reservation. Invalid or missing audio stays pending only +inside the bounded settlement window and becomes a visible terminal +operator-reconciliation item afterward. Recovery settles an abandoned capture; +it never resumes capture or renews recording authority after a process crash. + +Recovery requires durable shared access to both Postgres and `captureDir` +across pod replacement. A hard crash can discard audio written after the most +recent WAV checkpoint, so recovery intentionally may undercount by less than +the five-second checkpoint interval rather than bill unpersisted wall-clock +time. All hosted storage providers remain disabled in this branch, so no +reservation or audio capture can enter this path yet. + +## Rollout + +Start the worker with its normal Discord and provider credentials plus: + +```sh +export CALL_SCRIBE_HOSTED_CONTROL_PLANE_URL=https://callscribe.example.com +export CALL_SCRIBE_HOSTED_WORKLOAD_TOKEN=REDACTED +export CALL_SCRIBE_HOSTED_OUTBOX_ENCRYPTION_KEY=REDACTED_DIFFERENT_SECRET +export CALL_SCRIBE_HOSTED_WORKER_ID=worker-production-01 +cargo run --features discord -- discord --repo /meetings --skip-analysis +``` + +Use one active process for a Discord bot token; the database ownership fence is +defense in depth for rolling replacement and split ownership, not permission to +run two Discord consumers normally. Rotate the workload credential independently, +retain the previous credential only for a bounded overlap, and verify config +fetch, command acknowledgement, voice join/leave, retained audio, and usage +reconciliation before production enablement. diff --git a/docs/deployment/kubernetes.md b/docs/deployment/kubernetes.md index f214219..729da10 100644 --- a/docs/deployment/kubernetes.md +++ b/docs/deployment/kubernetes.md @@ -50,6 +50,11 @@ connectivity, storage, and the image. Scale to one replica only after any previous worker using the same Discord bot token is stopped. Do not run two workers with the same token. +For the optional SaaS adapter, set `hosted.enabled=true`, the HTTPS control +plane URL, and a dedicated Secret containing the workload token. See +[hosted-worker.md](hosted-worker.md) for the API contract and fail-closed gate. +The public chart never accepts Stripe credentials. + ## ZITADEL and API rollout order 1. Apply the reviewed declarative ZITADEL project, role, user grant, and diff --git a/migrations/20260812010000_hosted_worker_command_executions.sql b/migrations/20260812010000_hosted_worker_command_executions.sql new file mode 100644 index 0000000..df9e057 --- /dev/null +++ b/migrations/20260812010000_hosted_worker_command_executions.sql @@ -0,0 +1,46 @@ +-- Durable public-worker command state. The private control plane remains the +-- source of entitlement, billing, consent, and storage policy. + +CREATE TABLE IF NOT EXISTS call_scribe_hosted_command_executions ( + command_id TEXT PRIMARY KEY, + command_kind TEXT NOT NULL, + guild_id TEXT NOT NULL, + channel_id TEXT, + recording_notice_id TEXT, + generation BIGINT NOT NULL CHECK (generation >= 0), + status TEXT NOT NULL CHECK (status IN ('started', 'succeeded', 'failed')), + result JSONB, + error_code TEXT, + error_message TEXT, + started_at TIMESTAMPTZ NOT NULL DEFAULT now(), + completed_at TIMESTAMPTZ, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS call_scribe_hosted_command_executions_guild_idx + ON call_scribe_hosted_command_executions(guild_id, generation DESC); + +-- The lease token is application-encrypted with a dedicated stable key. It is +-- required only until the bounded server expiry and is never stored as text. +CREATE TABLE IF NOT EXISTS call_scribe_hosted_usage_outbox ( + reservation_id TEXT PRIMARY KEY, + encrypted_lease_token BYTEA NOT NULL, + encryption_nonce BYTEA NOT NULL, + recording_id TEXT NOT NULL, + actual_seconds BIGINT NOT NULL CHECK (actual_seconds > 0), + occurred_at TIMESTAMPTZ NOT NULL, + expires_at TIMESTAMPTZ NOT NULL, + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'delivered', 'expired')), + attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0), + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(), + last_error TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + delivered_at TIMESTAMPTZ, + CHECK (expires_at > occurred_at) +); + +CREATE INDEX IF NOT EXISTS call_scribe_hosted_usage_outbox_pending_idx + ON call_scribe_hosted_usage_outbox(next_attempt_at, expires_at) + WHERE status = 'pending'; diff --git a/migrations/20260812020000_hosted_capture_crash_recovery.sql b/migrations/20260812020000_hosted_capture_crash_recovery.sql new file mode 100644 index 0000000..2019c92 --- /dev/null +++ b/migrations/20260812020000_hosted_capture_crash_recovery.sql @@ -0,0 +1,83 @@ +-- Upgrade the already-deployed usage outbox so terminal rows can discard +-- encrypted lease material. This migration is intentionally additive: editing +-- the original CREATE TABLE would not alter an existing installation. +ALTER TABLE call_scribe_hosted_usage_outbox + ALTER COLUMN encrypted_lease_token DROP NOT NULL, + ALTER COLUMN encryption_nonce DROP NOT NULL; + +UPDATE call_scribe_hosted_usage_outbox +SET encrypted_lease_token = NULL, + encryption_nonce = NULL +WHERE status IN ('delivered', 'expired'); + +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conrelid = 'call_scribe_hosted_usage_outbox'::regclass + AND conname = 'call_scribe_hosted_usage_outbox_lease_state_check' + ) THEN + ALTER TABLE call_scribe_hosted_usage_outbox + ADD CONSTRAINT call_scribe_hosted_usage_outbox_lease_state_check + CHECK ( + (status = 'pending' + AND encrypted_lease_token IS NOT NULL + AND encryption_nonce IS NOT NULL) + OR (status IN ('delivered', 'expired') + AND encrypted_lease_token IS NULL + AND encryption_nonce IS NULL) + ) NOT VALID; + END IF; +END +$$; + +ALTER TABLE call_scribe_hosted_usage_outbox + VALIDATE CONSTRAINT call_scribe_hosted_usage_outbox_lease_state_check; + +-- A reservation must survive a process/pod crash before final WAV duration is +-- known. Each live process heartbeats its rows. Recovery may claim only a stale +-- owner and every mutation is fenced by the current recovery_claim_token. +-- Billable duration is derived from checkpointed mixed-audio WAV headers; +-- started_at is retained as evidence, never used as billable duration. +CREATE TABLE IF NOT EXISTS call_scribe_hosted_capture_recovery ( + reservation_id TEXT PRIMARY KEY, + encrypted_lease_token BYTEA, + encryption_nonce BYTEA, + recording_id TEXT NOT NULL UNIQUE, + base_wav_path TEXT NOT NULL, + reserved_seconds BIGINT NOT NULL CHECK (reserved_seconds > 0), + started_at TIMESTAMPTZ NOT NULL, + expires_at TIMESTAMPTZ NOT NULL, + owner_instance_id TEXT NOT NULL, + heartbeat_at TIMESTAMPTZ NOT NULL DEFAULT now(), + status TEXT NOT NULL DEFAULT 'active' + CHECK (status IN ('active', 'reconciling', 'expired')), + recovery_claim_token TEXT, + attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0), + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(), + recovery_lease_until TIMESTAMPTZ, + last_error TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CHECK (expires_at > started_at), + CHECK ( + (status = 'active' + AND encrypted_lease_token IS NOT NULL + AND encryption_nonce IS NOT NULL + AND recovery_claim_token IS NULL) + OR (status = 'reconciling' + AND encrypted_lease_token IS NOT NULL + AND encryption_nonce IS NOT NULL + AND recovery_claim_token IS NOT NULL + AND recovery_lease_until IS NOT NULL) + OR (status = 'expired' + AND encrypted_lease_token IS NULL + AND encryption_nonce IS NULL + AND recovery_claim_token IS NULL) + ) +); + +CREATE INDEX IF NOT EXISTS call_scribe_hosted_capture_recovery_claim_idx + ON call_scribe_hosted_capture_recovery(next_attempt_at, heartbeat_at, expires_at) + WHERE status IN ('active', 'reconciling'); diff --git a/src/hosted_control.rs b/src/hosted_control.rs new file mode 100644 index 0000000..0f8e35e --- /dev/null +++ b/src/hosted_control.rs @@ -0,0 +1,1329 @@ +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, RwLock}; +use std::time::{Duration, Instant}; + +use anyhow::{Context, Result, bail}; +use chacha20poly1305::aead::{Aead, KeyInit, Payload}; +use chacha20poly1305::{ChaCha20Poly1305, Nonce}; +use chrono::{DateTime, Utc}; +use reqwest::{Client, Method, StatusCode, Url, redirect}; +use serde::de::DeserializeOwned; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +const CONFIG_PATH: &str = "internal/v1/worker/guild-configurations"; +const COMMANDS_PATH: &str = "internal/v1/worker/commands"; +const USAGE_RESERVATIONS_PATH: &str = "internal/v1/worker/usage/reservations"; +const WORKER_ID_HEADER: &str = "X-Call-Scribe-Worker-Id"; +pub(crate) const RESERVATION_EXPIRY_MARGIN: Duration = Duration::from_secs(15); +pub(crate) const RESERVATION_MAX_LEASE: Duration = Duration::from_secs(90); +pub(crate) const RESERVATION_SETTLEMENT_GRACE: Duration = Duration::from_secs(30 * 60); +const RESERVATION_CLOCK_SKEW_TOLERANCE: Duration = Duration::from_secs(5); + +#[derive(Clone)] +pub(crate) struct HostedControlPlaneClient { + http: Client, + base_url: Url, + workload_token: String, + worker_id: String, + outbox_encryption_key: [u8; 32], +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub(crate) struct GuildConfiguration { + pub guild_id: String, + pub organization_id: String, + pub entitlement_active: bool, + #[serde(default)] + pub approved_channel_ids: Vec, + pub notice_channel_id: Option, + pub consent_mode: Option, + pub consent_policy_version: Option, + pub consent_notice_template: Option, + pub retention_days: Option, + pub recording_enabled: bool, + pub monthly_recording_seconds_cap: Option, + pub remaining_recording_seconds: Option, + pub storage_provider: Option, + pub storage_destination_label: Option, + pub ready: bool, + #[serde(default)] + pub blocked_reasons: Vec, + pub desired_recording_generation: u64, +} + +impl GuildConfiguration { + fn non_usage_controls_permit_recording(&self, channel_id: u64) -> bool { + self.entitlement_active + && self.recording_enabled + && !self.approved_channel_ids.is_empty() + && self + .approved_channel_ids + .iter() + .any(|approved| approved.parse::() == Ok(channel_id)) + && self + .notice_channel_id + .as_deref() + .is_some_and(|id| id.parse::().is_ok_and(|id| id > 0)) + && self + .consent_mode + .as_deref() + .is_some_and(|mode| mode == "explicit_command") + && self + .consent_policy_version + .as_deref() + .is_some_and(|version| !version.trim().is_empty()) + && self + .consent_notice_template + .as_deref() + .is_some_and(|notice| (20..=1_500).contains(¬ice.chars().count())) + && self + .retention_days + .is_some_and(|days| (1..=365).contains(&days)) + && self + .monthly_recording_seconds_cap + .is_some_and(|cap| cap > 0) + } + + fn core_controls_permit_recording(&self, channel_id: u64) -> bool { + self.non_usage_controls_permit_recording(channel_id) + && self + .remaining_recording_seconds + .is_some_and(|remaining| remaining > 0) + && self.remaining_recording_seconds <= self.monthly_recording_seconds_cap + && self.ready + && self.blocked_reasons.is_empty() + } + + fn core_controls_permit_continuation(&self, channel_id: u64) -> bool { + let only_current_reservation_exhausted_usage = !self.blocked_reasons.is_empty() + && self + .blocked_reasons + .iter() + .all(|reason| reason == "usage_cap_exhausted"); + self.non_usage_controls_permit_recording(channel_id) + && self.remaining_recording_seconds <= self.monthly_recording_seconds_cap + && ((self.ready && self.blocked_reasons.is_empty()) + || (!self.ready && only_current_reservation_exhausted_usage)) + } + + pub(crate) fn permits_recording(&self, channel_id: u64) -> bool { + self.core_controls_permit_recording(channel_id) + && storage_destination_supported(self.storage_provider.as_deref()) + } + + pub(crate) fn permits_continuation(&self, channel_id: u64) -> bool { + self.core_controls_permit_continuation(channel_id) + && storage_destination_supported(self.storage_provider.as_deref()) + } +} + +#[derive(Clone, Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct GuildConfigurationResponse { + pub revision: String, + #[serde(default)] + pub guilds: Vec, +} + +#[derive(Clone, Debug, Default)] +struct Snapshot { + revision: Option, + fetched_at: Option, + guilds: HashMap, +} + +#[derive(Clone, Debug)] +pub(crate) struct HostedConfigurationStore { + max_staleness: Duration, + snapshot: Arc>, +} + +impl HostedConfigurationStore { + pub(crate) fn new(max_staleness: Duration) -> Self { + Self { + max_staleness, + snapshot: Arc::new(RwLock::new(Snapshot::default())), + } + } + + pub(crate) fn replace(&self, response: GuildConfigurationResponse) -> HashSet { + let mut guilds = HashMap::new(); + for guild in response.guilds { + if let Ok(guild_id) = guild.guild_id.parse::() { + guilds.insert(guild_id, guild); + } + } + + let mut snapshot = self + .snapshot + .write() + .expect("hosted configuration lock poisoned"); + let changed_guilds = snapshot + .guilds + .keys() + .chain(guilds.keys()) + .copied() + .collect(); + snapshot.revision = Some(response.revision); + snapshot.fetched_at = Some(Instant::now()); + snapshot.guilds = guilds; + changed_guilds + } + + pub(crate) fn guild_ids(&self) -> Vec { + self.snapshot + .read() + .expect("hosted configuration lock poisoned") + .guilds + .keys() + .copied() + .collect() + } + + pub(crate) fn policy_for(&self, guild_id: u64) -> Option { + let snapshot = self + .snapshot + .read() + .expect("hosted configuration lock poisoned"); + if snapshot + .fetched_at + .is_none_or(|fetched_at| fetched_at.elapsed() > self.max_staleness) + { + return None; + } + snapshot.guilds.get(&guild_id).cloned() + } +} + +#[derive(Clone, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub(crate) struct WorkerCommand { + pub id: String, + pub command_kind: String, + pub guild_id: String, + pub channel_id: Option, + pub lease_token: String, + pub lease_expires_at: DateTime, + pub generation: u64, + pub recording_notice_id: Option, +} + +#[derive(Clone, Deserialize)] +pub(crate) struct WorkerCommandsResponse { + #[serde(default)] + pub commands: Vec, +} + +#[derive(Clone, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub(crate) struct UsageReservation { + pub reservation_id: String, + pub lease_token: String, + pub reserved_seconds: u64, + pub expires_at: String, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct CommandLeaseRequest<'a> { + worker_id: &'a str, + limit: u8, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct CommandCompletion<'a> { + lease_token: &'a str, + success: bool, + result: serde_json::Value, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct UsageReservationRequest<'a> { + command_id: &'a str, + requested_seconds: u64, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct UsageConsumption<'a> { + lease_token: &'a str, + recording_id: &'a str, + actual_seconds: u64, + occurred_at: &'a str, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct UsageRelease<'a> { + lease_token: &'a str, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct UsageHeartbeat<'a> { + lease_token: &'a str, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct UsageHeartbeatResponse { + reservation_id: String, + expires_at: String, +} + +impl HostedControlPlaneClient { + pub(crate) fn new( + base_url: &str, + workload_token: String, + worker_id: String, + outbox_encryption_secret: String, + ) -> Result { + Self::new_with_timeout( + base_url, + workload_token, + worker_id, + outbox_encryption_secret, + Duration::from_secs(10), + ) + } + + fn new_with_timeout( + base_url: &str, + workload_token: String, + worker_id: String, + outbox_encryption_secret: String, + request_timeout: Duration, + ) -> Result { + if workload_token.len() < 32 || workload_token.trim() != workload_token { + bail!("hosted workload token must contain at least 32 bytes"); + } + if outbox_encryption_secret.len() < 32 + || outbox_encryption_secret.trim() != outbox_encryption_secret + || outbox_encryption_secret == workload_token + { + bail!( + "hosted outbox encryption secret must contain at least 32 bytes and differ from the workload token" + ); + } + if worker_id.is_empty() + || worker_id.len() > 128 + || !worker_id + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.')) + { + bail!("hosted worker id must be 1-128 ASCII identifier characters"); + } + let mut base_url = Url::parse(base_url).context("invalid hosted control-plane URL")?; + if base_url.scheme() != "https" && !base_url.host_str().is_some_and(is_loopback_host) { + bail!("hosted control-plane URL must use HTTPS (except loopback development)"); + } + if !base_url.username().is_empty() + || base_url.password().is_some() + || base_url.query().is_some() + || base_url.fragment().is_some() + || base_url.path() != "/" + { + bail!( + "hosted control-plane URL must be an origin without credentials, path, query, or fragment" + ); + } + base_url.set_path("/"); + if request_timeout.is_zero() { + bail!("hosted control-plane request timeout must be positive"); + } + let http = Client::builder() + .timeout(request_timeout) + .redirect(redirect::Policy::none()) + .build() + .context("failed to build hosted control-plane HTTP client")?; + let mut key_derivation = Sha256::new(); + key_derivation.update(b"call-scribe-hosted-usage-outbox-v1\0"); + key_derivation.update(outbox_encryption_secret.as_bytes()); + let outbox_encryption_key: [u8; 32] = key_derivation.finalize().into(); + Ok(Self { + http, + base_url, + workload_token, + worker_id, + outbox_encryption_key, + }) + } + + pub(crate) fn encrypt_reservation_lease( + &self, + reservation_id: &str, + lease_token: &str, + ) -> Result<(Vec, Vec)> { + let cipher = ChaCha20Poly1305::new_from_slice(&self.outbox_encryption_key) + .expect("hosted outbox key has the required length"); + let nonce_bytes: [u8; 12] = rand::random(); + let ciphertext = cipher + .encrypt( + Nonce::from_slice(&nonce_bytes), + Payload { + msg: lease_token.as_bytes(), + aad: reservation_id.as_bytes(), + }, + ) + .map_err(|_| anyhow::anyhow!("failed to encrypt hosted reservation lease"))?; + Ok((ciphertext, nonce_bytes.to_vec())) + } + + pub(crate) fn decrypt_reservation_lease( + &self, + reservation_id: &str, + ciphertext: &[u8], + nonce: &[u8], + ) -> Result { + if nonce.len() != 12 { + bail!("hosted reservation lease nonce was invalid"); + } + let cipher = ChaCha20Poly1305::new_from_slice(&self.outbox_encryption_key) + .expect("hosted outbox key has the required length"); + let plaintext = cipher + .decrypt( + Nonce::from_slice(nonce), + Payload { + msg: ciphertext, + aad: reservation_id.as_bytes(), + }, + ) + .map_err(|_| anyhow::anyhow!("failed to decrypt hosted reservation lease"))?; + String::from_utf8(plaintext).context("hosted reservation lease was not valid UTF-8") + } + + fn authenticated_request(&self, method: Method, url: Url) -> reqwest::RequestBuilder { + self.http + .request(method, url) + .bearer_auth(&self.workload_token) + .header(WORKER_ID_HEADER, &self.worker_id) + } + + pub(crate) async fn fetch_configurations(&self) -> Result { + let url = self.base_url.join(CONFIG_PATH)?; + let response = self + .authenticated_request(Method::GET, url) + .send() + .await + .context("hosted configuration request failed")?; + if response.status() != StatusCode::OK { + bail!( + "hosted configuration request returned HTTP {}", + response.status() + ); + } + let response: GuildConfigurationResponse = decode_json_bounded(response, 2 * 1024 * 1024) + .await + .context("hosted configuration response was invalid")?; + validate_configuration_response(&response)?; + Ok(response) + } + + /// Durable command polling boundary. Hosted control planes can enqueue + /// explicit start/stop commands; the worker never infers a hosted start + /// merely from channel occupancy. + pub(crate) async fn lease_commands(&self) -> Result { + let url = self.base_url.join(&format!("{COMMANDS_PATH}/lease"))?; + let response = self + .authenticated_request(Method::POST, url) + .json(&CommandLeaseRequest { + worker_id: &self.worker_id, + limit: 10, + }) + .send() + .await + .context("hosted command request failed")?; + if response.status() != StatusCode::OK { + bail!("hosted command request returned HTTP {}", response.status()); + } + let response: WorkerCommandsResponse = decode_json_bounded(response, 256 * 1024) + .await + .context("hosted command response was invalid")?; + if response.commands.len() > 10 { + bail!("hosted command response exceeded the requested limit"); + } + for command in &response.commands { + validate_worker_command(command)?; + } + Ok(response) + } + + pub(crate) async fn complete_command( + &self, + command_id: &str, + lease_token: &str, + success: bool, + result: serde_json::Value, + ) -> Result<()> { + validate_command_result(&result)?; + let path = format!( + "{COMMANDS_PATH}/{}/complete", + urlencoding::encode(command_id) + ); + let response = self + .authenticated_request(Method::POST, self.base_url.join(&path)?) + .json(&CommandCompletion { + lease_token, + success, + result, + }) + .send() + .await + .context("hosted command acknowledgement failed")?; + if !response.status().is_success() { + bail!( + "hosted command acknowledgement returned HTTP {}", + response.status() + ); + } + Ok(()) + } + + pub(crate) async fn reserve_usage( + &self, + command_id: &str, + requested_seconds: u64, + ) -> Result { + if command_id.is_empty() + || command_id.len() > 200 + || !(1..=3_600).contains(&requested_seconds) + { + bail!("hosted usage reservation request was invalid"); + } + let response = self + .authenticated_request(Method::POST, self.base_url.join(USAGE_RESERVATIONS_PATH)?) + .json(&UsageReservationRequest { + command_id, + requested_seconds, + }) + .send() + .await + .context("hosted usage reservation failed")?; + if response.status() != StatusCode::OK && response.status() != StatusCode::CREATED { + bail!( + "hosted usage reservation returned HTTP {}", + response.status() + ); + } + let reservation: UsageReservation = decode_json_bounded(response, 64 * 1024) + .await + .context("hosted usage reservation response was invalid")?; + if reservation.reservation_id.is_empty() + || reservation.lease_token.is_empty() + || reservation.reserved_seconds == 0 + || reservation.reserved_seconds > requested_seconds + { + bail!("hosted usage reservation response was incomplete"); + } + let expires_at = DateTime::parse_from_rfc3339(&reservation.expires_at) + .context("hosted usage reservation expiry was invalid")? + .with_timezone(&Utc); + validate_reservation_expiry(expires_at, "reservation")?; + Ok(reservation) + } + + pub(crate) async fn consume_usage( + &self, + reservation: &UsageReservation, + recording_id: &str, + actual_seconds: u64, + occurred_at: &str, + ) -> Result<()> { + let path = format!( + "{USAGE_RESERVATIONS_PATH}/{}/consume", + urlencoding::encode(&reservation.reservation_id) + ); + let response = self + .authenticated_request(Method::POST, self.base_url.join(&path)?) + .json(&UsageConsumption { + lease_token: &reservation.lease_token, + recording_id, + actual_seconds, + occurred_at, + }) + .send() + .await + .context("hosted usage consumption failed")?; + if !response.status().is_success() { + bail!( + "hosted usage consumption returned HTTP {}", + response.status() + ); + } + Ok(()) + } + + pub(crate) async fn heartbeat_usage( + &self, + reservation: &UsageReservation, + ) -> Result> { + let path = format!( + "{USAGE_RESERVATIONS_PATH}/{}/heartbeat", + urlencoding::encode(&reservation.reservation_id) + ); + let response = self + .authenticated_request(Method::POST, self.base_url.join(&path)?) + .json(&UsageHeartbeat { + lease_token: &reservation.lease_token, + }) + .send() + .await + .context("hosted usage heartbeat failed")?; + if response.status() != StatusCode::OK { + bail!("hosted usage heartbeat returned HTTP {}", response.status()); + } + let heartbeat: UsageHeartbeatResponse = decode_json_bounded(response, 64 * 1024) + .await + .context("hosted usage heartbeat response was invalid")?; + if heartbeat.reservation_id != reservation.reservation_id { + bail!("hosted usage heartbeat returned a mismatched reservation id"); + } + let expires_at = DateTime::parse_from_rfc3339(&heartbeat.expires_at) + .context("hosted usage heartbeat expiry was invalid")? + .with_timezone(&Utc); + validate_reservation_expiry(expires_at, "heartbeat")?; + Ok(expires_at) + } + + pub(crate) async fn release_usage(&self, reservation: &UsageReservation) -> Result<()> { + let path = format!( + "{USAGE_RESERVATIONS_PATH}/{}/release", + urlencoding::encode(&reservation.reservation_id) + ); + let response = self + .authenticated_request(Method::POST, self.base_url.join(&path)?) + .json(&UsageRelease { + lease_token: &reservation.lease_token, + }) + .send() + .await + .context("hosted usage release failed")?; + if !response.status().is_success() { + bail!("hosted usage release returned HTTP {}", response.status()); + } + Ok(()) + } +} + +fn validate_reservation_expiry(expires_at: DateTime, source: &str) -> Result<()> { + let now = Utc::now(); + let minimum_expiry = now + + chrono::Duration::from_std(RESERVATION_EXPIRY_MARGIN) + .expect("reservation expiry margin must fit chrono duration"); + if expires_at <= minimum_expiry { + bail!("hosted usage {source} expires too soon"); + } + let maximum_expiry = now + + chrono::Duration::from_std(RESERVATION_MAX_LEASE + RESERVATION_CLOCK_SKEW_TOLERANCE) + .expect("reservation maximum lease must fit chrono duration"); + if expires_at > maximum_expiry { + bail!("hosted usage {source} expiry exceeded the lease contract"); + } + Ok(()) +} + +fn validate_command_result(result: &serde_json::Value) -> Result<()> { + const ALLOWED_KEYS: &[&str] = &["code", "message", "recordingId", "durationSeconds"]; + let object = result + .as_object() + .context("hosted command result must be an object")?; + if object + .keys() + .any(|key| !ALLOWED_KEYS.contains(&key.as_str())) + { + bail!("hosted command result included an unsupported field"); + } + for (key, value) in object { + let valid_type = match key.as_str() { + "code" | "message" | "recordingId" => value.is_string(), + "durationSeconds" => value.as_u64().is_some(), + _ => false, + }; + if !valid_type { + bail!("hosted command result included an invalid field value"); + } + } + let encoded = serde_json::to_vec(result)?; + if encoded.len() > 1_024 { + bail!("hosted command result exceeded 1024 bytes"); + } + Ok(()) +} + +async fn decode_json_bounded( + mut response: reqwest::Response, + maximum_bytes: usize, +) -> Result { + if response + .content_length() + .is_some_and(|length| length > maximum_bytes as u64) + { + bail!("control-plane response exceeded the maximum size"); + } + let mut body = Vec::new(); + while let Some(chunk) = response + .chunk() + .await + .context("failed to read control-plane response")? + { + if body.len().saturating_add(chunk.len()) > maximum_bytes { + bail!("control-plane response exceeded the maximum size"); + } + body.extend_from_slice(&chunk); + } + serde_json::from_slice(&body).context("control-plane response was not valid JSON") +} + +fn validate_configuration_response(response: &GuildConfigurationResponse) -> Result<()> { + if response.revision.is_empty() || response.revision.len() > 128 { + bail!("hosted configuration revision was invalid"); + } + if response.guilds.len() > 10_000 { + bail!("hosted configuration included too many guilds"); + } + let mut guild_ids = HashSet::new(); + for guild in &response.guilds { + let guild_id = guild + .guild_id + .parse::() + .context("hosted configuration guild id was invalid")?; + if guild_id == 0 || !guild_ids.insert(guild_id) { + bail!("hosted configuration guild id was zero or duplicated"); + } + if guild.organization_id.is_empty() || guild.organization_id.len() > 128 { + bail!("hosted configuration organization id was invalid"); + } + if guild.approved_channel_ids.len() > 1_000 + || guild.approved_channel_ids.iter().any(|channel_id| { + !channel_id + .parse::() + .is_ok_and(|channel_id| channel_id > 0) + }) + { + bail!("hosted configuration approved channel ids were invalid"); + } + } + Ok(()) +} + +fn validate_worker_command(command: &WorkerCommand) -> Result<()> { + if command.id.is_empty() + || command.id.len() > 200 + || command.lease_token.len() < 16 + || command.lease_token.len() > 1_024 + || !command + .guild_id + .parse::() + .is_ok_and(|guild_id| guild_id > 0) + || command.lease_expires_at <= Utc::now() + { + bail!("hosted worker command identifiers were invalid"); + } + if !matches!( + command.command_kind.as_str(), + "record_start" | "record_stop" + ) { + bail!("hosted worker command kind was unsupported"); + } + if command.command_kind == "record_start" + && (!command + .recording_notice_id + .as_deref() + .is_some_and(|notice_id| !notice_id.trim().is_empty() && notice_id.len() <= 200) + || command + .channel_id + .as_deref() + .and_then(|channel_id| channel_id.parse::().ok()) + .is_none_or(|channel_id| channel_id == 0)) + { + bail!("hosted start command omitted valid channel or recording notice evidence"); + } + Ok(()) +} + +fn is_loopback_host(host: &str) -> bool { + matches!(host, "localhost" | "127.0.0.1" | "[::1]" | "::1") +} + +fn storage_destination_supported(provider: Option<&str>) -> bool { + // A provider is only enabled after this crate can upload the completed + // artifacts, verify delivery, and apply local deletion/retention. Keeping + // this exhaustive and fail-closed prevents control-plane configuration + // from silently degrading to local PVC retention. + match provider { + Some("customer_s3" | "customer_r2" | "google_drive" | "managed_transient") | None => false, + Some(_) => false, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::extract::Path; + use axum::http::HeaderMap; + use axum::response::IntoResponse; + use axum::routing::post; + use axum::{Json, Router}; + + fn reservation(expires_at: DateTime) -> UsageReservation { + UsageReservation { + reservation_id: "reservation-1".to_string(), + lease_token: "opaque-lease-token".to_string(), + reserved_seconds: 300, + expires_at: expires_at.to_rfc3339(), + } + } + + fn test_client(base_url: &str, timeout: Duration) -> HostedControlPlaneClient { + HostedControlPlaneClient::new_with_timeout( + base_url, + "test-secret-with-at-least-thirty-two-bytes".to_string(), + "worker-1".to_string(), + "test-outbox-secret-with-at-least-thirty-two-bytes".to_string(), + timeout, + ) + .expect("loopback hosted client should be valid") + } + + async fn spawn_heartbeat_server( + status: StatusCode, + response_body: String, + delay: Duration, + ) -> String { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("test listener should bind"); + let address = listener + .local_addr() + .expect("test listener should have an address"); + let app = Router::new().route( + "/internal/v1/worker/usage/reservations/{reservation_id}/heartbeat", + post( + move |Path(reservation_id): Path, + headers: HeaderMap, + Json(input): Json| { + let response_body = response_body.clone(); + async move { + tokio::time::sleep(delay).await; + if reservation_id != "reservation-1" + || input != serde_json::json!({"leaseToken": "opaque-lease-token"}) + || headers + .get(WORKER_ID_HEADER) + .and_then(|value| value.to_str().ok()) + != Some("worker-1") + || headers + .get(reqwest::header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + != Some("Bearer test-secret-with-at-least-thirty-two-bytes") + { + return StatusCode::UNAUTHORIZED.into_response(); + } + ( + status, + [(reqwest::header::CONTENT_TYPE, "application/json")], + response_body, + ) + .into_response() + } + }, + ), + ); + tokio::spawn(async move { + axum::serve(listener, app) + .await + .expect("test heartbeat server should run"); + }); + format!("http://{address}") + } + + fn enabled_policy() -> GuildConfiguration { + GuildConfiguration { + guild_id: "1".to_string(), + organization_id: "org_1".to_string(), + entitlement_active: true, + approved_channel_ids: vec!["2".to_string()], + notice_channel_id: Some("3".to_string()), + consent_mode: Some("explicit_command".to_string()), + consent_policy_version: Some("v1".to_string()), + consent_notice_template: Some( + "This call will be recorded after a durable start command.".to_string(), + ), + retention_days: Some(30), + recording_enabled: true, + monthly_recording_seconds_cap: Some(3_600), + remaining_recording_seconds: Some(30), + storage_provider: Some("customer_s3".to_string()), + storage_destination_label: Some("worker-pvc".to_string()), + ready: true, + blocked_reasons: Vec::new(), + desired_recording_generation: 1, + } + } + + #[test] + fn core_controls_require_fully_authorized_approved_channel() { + let policy = enabled_policy(); + assert!(policy.core_controls_permit_recording(2)); + assert!(!policy.permits_recording(2)); + assert!(!policy.core_controls_permit_recording(9)); + } + + #[test] + fn missing_or_exhausted_controls_fail_closed() { + let mut policy = enabled_policy(); + policy.entitlement_active = false; + assert!(!policy.core_controls_permit_recording(2)); + policy.entitlement_active = true; + policy.notice_channel_id = None; + assert!(!policy.core_controls_permit_recording(2)); + policy.notice_channel_id = Some("3".to_string()); + policy.remaining_recording_seconds = Some(0); + assert!(!policy.core_controls_permit_recording(2)); + policy.remaining_recording_seconds = None; + assert!(!policy.core_controls_permit_recording(2)); + policy.remaining_recording_seconds = Some(30); + policy.retention_days = None; + assert!(!policy.core_controls_permit_recording(2)); + policy.retention_days = Some(30); + policy.monthly_recording_seconds_cap = Some(20); + assert!(!policy.core_controls_permit_recording(2)); + } + + #[test] + fn matching_reservation_can_continue_only_through_usage_exhaustion() { + let mut policy = enabled_policy(); + policy.remaining_recording_seconds = Some(0); + policy.ready = false; + policy.blocked_reasons = vec!["usage_cap_exhausted".to_string()]; + + assert!(!policy.core_controls_permit_recording(2)); + assert!(policy.core_controls_permit_continuation(2)); + + policy.entitlement_active = false; + assert!(!policy.core_controls_permit_continuation(2)); + policy.entitlement_active = true; + policy + .blocked_reasons + .push("privacy_restricted".to_string()); + assert!(!policy.core_controls_permit_continuation(2)); + } + + #[test] + fn store_has_no_policy_before_first_refresh() { + let store = HostedConfigurationStore::new(Duration::from_secs(30)); + assert!(store.policy_for(1).is_none()); + } + + #[test] + fn zero_staleness_expires_a_refreshed_policy() { + let store = HostedConfigurationStore::new(Duration::from_millis(1)); + store.replace(GuildConfigurationResponse { + revision: "r1".to_string(), + guilds: vec![enabled_policy()], + }); + std::thread::sleep(Duration::from_millis(2)); + assert!(store.policy_for(1).is_none()); + } + + #[test] + fn malformed_guild_ids_are_ignored() { + let store = HostedConfigurationStore::new(Duration::from_secs(30)); + let mut invalid = enabled_policy(); + invalid.guild_id = "not-an-id".to_string(); + store.replace(GuildConfigurationResponse { + revision: "r1".to_string(), + guilds: vec![invalid, enabled_policy()], + }); + assert_eq!(store.guild_ids(), vec![1]); + } + + #[test] + fn production_control_plane_requires_https() { + assert!( + HostedControlPlaneClient::new( + "http://control.example.com", + "test-secret-with-at-least-thirty-two-bytes".to_string(), + "worker-1".to_string(), + "test-outbox-secret-with-at-least-thirty-two-bytes".to_string() + ) + .is_err() + ); + assert!( + HostedControlPlaneClient::new( + "http://127.0.0.1:8080", + "test-secret-with-at-least-thirty-two-bytes".to_string(), + "worker-1".to_string(), + "test-outbox-secret-with-at-least-thirty-two-bytes".to_string() + ) + .is_ok() + ); + assert!( + HostedControlPlaneClient::new( + "https://control.example.com/path", + "test-secret-with-at-least-thirty-two-bytes".to_string(), + "worker-1".to_string(), + "test-outbox-secret-with-at-least-thirty-two-bytes".to_string() + ) + .is_err() + ); + assert!( + HostedControlPlaneClient::new( + "https://control.example.com", + "short".to_string(), + "worker-1".to_string(), + "test-outbox-secret-with-at-least-thirty-two-bytes".to_string() + ) + .is_err() + ); + } + + #[test] + fn authenticated_requests_bind_bearer_and_worker_id() { + let client = HostedControlPlaneClient::new( + "http://127.0.0.1:8080", + "test-secret-with-at-least-thirty-two-bytes".to_string(), + "worker-1".to_string(), + "test-outbox-secret-with-at-least-thirty-two-bytes".to_string(), + ) + .expect("loopback client should be valid"); + let request = client + .authenticated_request( + Method::GET, + client + .base_url + .join(CONFIG_PATH) + .expect("configuration URL should join"), + ) + .build() + .expect("authenticated request should build"); + assert_eq!( + request + .headers() + .get(WORKER_ID_HEADER) + .and_then(|value| value.to_str().ok()), + Some("worker-1") + ); + assert_eq!( + request + .headers() + .get(reqwest::header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()), + Some("Bearer test-secret-with-at-least-thirty-two-bytes") + ); + } + + #[test] + fn command_results_are_allowlisted_and_bounded() { + assert!( + validate_command_result(&serde_json::json!({ + "code": "recording_started", + "recordingId": "recording-1" + })) + .is_ok() + ); + assert!( + validate_command_result(&serde_json::json!({ + "secret": "must-not-be-persisted" + })) + .is_err() + ); + assert!( + validate_command_result(&serde_json::json!({ + "message": "x".repeat(2_000) + })) + .is_err() + ); + assert!( + validate_command_result(&serde_json::json!({ + "durationSeconds": "not-a-number" + })) + .is_err() + ); + } + + #[test] + fn reservation_lease_encryption_is_bound_to_reservation() { + let client = HostedControlPlaneClient::new( + "http://127.0.0.1:8080", + "test-secret-with-at-least-thirty-two-bytes".to_string(), + "worker-1".to_string(), + "test-outbox-secret-with-at-least-thirty-two-bytes".to_string(), + ) + .expect("loopback client should be valid"); + let (ciphertext, nonce) = client + .encrypt_reservation_lease("reservation-1", "opaque-lease-token") + .expect("lease should encrypt"); + assert_ne!(ciphertext, b"opaque-lease-token"); + assert_eq!( + client + .decrypt_reservation_lease("reservation-1", &ciphertext, &nonce) + .expect("lease should decrypt"), + "opaque-lease-token" + ); + assert!( + client + .decrypt_reservation_lease("reservation-2", &ciphertext, &nonce) + .is_err() + ); + } + + #[test] + fn configuration_response_rejects_duplicate_guilds() { + let response = GuildConfigurationResponse { + revision: "r1".to_string(), + guilds: vec![enabled_policy(), enabled_policy()], + }; + assert!(validate_configuration_response(&response).is_err()); + } + + #[test] + fn control_plane_configuration_wire_shape_deserializes() { + let response: GuildConfigurationResponse = serde_json::from_value(serde_json::json!({ + "revision": "abc123", + "generatedAt": "2026-08-12T18:00:00Z", + "guilds": [{ + "guildId": "123", + "organizationId": "019fa65e-d054-7582-8f88-a8ed258adf0f", + "entitlementActive": true, + "approvedChannelIds": ["456"], + "noticeChannelId": "789", + "consentMode": "explicit_command", + "consentPolicyVersion": "v1", + "consentNoticeTemplate": "This call is being recorded with advance notice.", + "retentionDays": 30, + "recordingEnabled": true, + "monthlyRecordingSecondsCap": 3600, + "remainingRecordingSeconds": 3600, + "storageProvider": "customer_s3", + "storageDestinationLabel": "Customer archive", + "ready": true, + "blockedReasons": [], + "desiredRecordingGeneration": 4 + }] + })) + .expect("control-plane snapshot should deserialize"); + assert!(validate_configuration_response(&response).is_ok()); + assert!(!response.guilds[0].permits_recording(456)); + } + + #[test] + fn command_contract_accepts_leased_start() { + let command: WorkerCommand = serde_json::from_value(serde_json::json!({ + "id": "019fa65e-d054-7582-8f88-a8ed258adf0f", + "commandKind": "record_start", + "guildId": "1", + "channelId": "2", + "leaseToken": "opaque-lease-token", + "leaseExpiresAt": "2099-01-01T00:00:00Z", + "generation": 1, + "recordingNoticeId": "019fa65e-d054-7582-8f88-a8ed258adf10" + })) + .expect("command should deserialize"); + assert!(validate_worker_command(&command).is_ok()); + } + + #[test] + fn reservation_contract_uses_command_bound_wire_shape() { + let request = UsageReservationRequest { + command_id: "cmd_01", + requested_seconds: 3_600, + }; + assert_eq!( + serde_json::to_value(request).expect("reservation request should serialize"), + serde_json::json!({ + "commandId": "cmd_01", + "requestedSeconds": 3_600 + }) + ); + } + + #[tokio::test] + async fn heartbeat_accepts_authoritative_renewal_and_shortening() { + let initial_expiry = Utc::now() + chrono::Duration::seconds(45); + let renewed_expiry = Utc::now() + chrono::Duration::seconds(80); + let server = spawn_heartbeat_server( + StatusCode::OK, + serde_json::json!({ + "reservationId": "reservation-1", + "expiresAt": renewed_expiry.to_rfc3339(), + }) + .to_string(), + Duration::ZERO, + ) + .await; + let renewed = test_client(&server, Duration::from_secs(1)) + .heartbeat_usage(&reservation(initial_expiry)) + .await + .expect("a valid renewal should be accepted"); + assert_eq!(renewed, renewed_expiry); + + let prior_expiry = Utc::now() + chrono::Duration::seconds(80); + let shortened_expiry = Utc::now() + chrono::Duration::seconds(45); + let server = spawn_heartbeat_server( + StatusCode::OK, + serde_json::json!({ + "reservationId": "reservation-1", + "expiresAt": shortened_expiry.to_rfc3339(), + }) + .to_string(), + Duration::ZERO, + ) + .await; + let shortened = test_client(&server, Duration::from_secs(1)) + .heartbeat_usage(&reservation(prior_expiry)) + .await + .expect("a still-safe authoritative shortening should be accepted"); + assert_eq!(shortened, shortened_expiry); + } + + #[tokio::test] + async fn heartbeat_rejects_mismatch_malformed_and_near_expiry() { + let valid_reservation = reservation(Utc::now() + chrono::Duration::seconds(80)); + let mismatch_server = spawn_heartbeat_server( + StatusCode::OK, + serde_json::json!({ + "reservationId": "reservation-2", + "expiresAt": (Utc::now() + chrono::Duration::seconds(80)).to_rfc3339(), + }) + .to_string(), + Duration::ZERO, + ) + .await; + assert!( + test_client(&mismatch_server, Duration::from_secs(1)) + .heartbeat_usage(&valid_reservation) + .await + .is_err() + ); + + let malformed_server = spawn_heartbeat_server( + StatusCode::OK, + r#"{"reservationId":"reservation-1"}"#.to_string(), + Duration::ZERO, + ) + .await; + assert!( + test_client(&malformed_server, Duration::from_secs(1)) + .heartbeat_usage(&valid_reservation) + .await + .is_err() + ); + + let near_expiry_server = spawn_heartbeat_server( + StatusCode::OK, + serde_json::json!({ + "reservationId": "reservation-1", + "expiresAt": (Utc::now() + chrono::Duration::seconds(5)).to_rfc3339(), + }) + .to_string(), + Duration::ZERO, + ) + .await; + assert!( + test_client(&near_expiry_server, Duration::from_secs(1)) + .heartbeat_usage(&valid_reservation) + .await + .is_err() + ); + } + + #[tokio::test] + async fn heartbeat_rejects_auth_revocation_timeout_and_outage() { + let valid_reservation = reservation(Utc::now() + chrono::Duration::seconds(80)); + let auth_server = spawn_heartbeat_server( + StatusCode::UNAUTHORIZED, + r#"{"error":"workload auth revoked"}"#.to_string(), + Duration::ZERO, + ) + .await; + assert!( + test_client(&auth_server, Duration::from_secs(1)) + .heartbeat_usage(&valid_reservation) + .await + .is_err() + ); + + let revoked_server = spawn_heartbeat_server( + StatusCode::CONFLICT, + r#"{"error":"reservation revoked"}"#.to_string(), + Duration::ZERO, + ) + .await; + assert!( + test_client(&revoked_server, Duration::from_secs(1)) + .heartbeat_usage(&valid_reservation) + .await + .is_err() + ); + + let slow_server = spawn_heartbeat_server( + StatusCode::OK, + serde_json::json!({ + "reservationId": "reservation-1", + "expiresAt": (Utc::now() + chrono::Duration::seconds(80)).to_rfc3339(), + }) + .to_string(), + Duration::from_millis(100), + ) + .await; + assert!( + test_client(&slow_server, Duration::from_millis(20)) + .heartbeat_usage(&valid_reservation) + .await + .is_err() + ); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("outage listener should bind"); + let address = listener + .local_addr() + .expect("outage listener should have an address"); + drop(listener); + assert!( + test_client(&format!("http://{address}"), Duration::from_secs(1)) + .heartbeat_usage(&valid_reservation) + .await + .is_err() + ); + } + + #[tokio::test] + async fn heartbeat_rejects_expiry_beyond_contract_horizon() { + let valid_reservation = reservation(Utc::now() + chrono::Duration::seconds(80)); + let server = spawn_heartbeat_server( + StatusCode::OK, + serde_json::json!({ + "reservationId": "reservation-1", + "expiresAt": (Utc::now() + chrono::Duration::seconds(120)).to_rfc3339(), + }) + .to_string(), + Duration::ZERO, + ) + .await; + assert!( + test_client(&server, Duration::from_secs(1)) + .heartbeat_usage(&valid_reservation) + .await + .is_err() + ); + } + + #[test] + fn all_current_hosted_storage_providers_fail_closed() { + for provider in [ + None, + Some("customer_s3"), + Some("customer_r2"), + Some("google_drive"), + Some("managed_transient"), + Some("unknown"), + ] { + assert!(!storage_destination_supported(provider)); + } + } +} diff --git a/src/main.rs b/src/main.rs index b0a180d..8dce124 100644 --- a/src/main.rs +++ b/src/main.rs @@ -10,7 +10,7 @@ use std::{ io::BufWriter, num::NonZeroU8, ops::DerefMut, - sync::atomic::{AtomicU32, AtomicU64, Ordering}, + sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}, sync::{Arc, Mutex}, }; @@ -23,9 +23,16 @@ use clap::{Parser, ValueEnum}; use dashmap::DashMap; mod api; mod github_issues; +#[cfg(feature = "discord")] +mod hosted_control; mod oidc_session; mod providers; +#[cfg(feature = "discord")] +use hosted_control::{ + HostedConfigurationStore, HostedControlPlaneClient, RESERVATION_EXPIRY_MARGIN, + RESERVATION_SETTLEMENT_GRACE, UsageReservation, WorkerCommand, +}; use providers::{ ElevenLabsSttConfig, ElevenLabsSttProvider, OpenAiSttConfig, OpenAiSttProvider, SpeechToTextProvider, TranscriptionRequest, TranscriptionResponse, transcribe, @@ -81,9 +88,19 @@ const DISCORD_WAV_SEGMENT_MAX_BYTES: u32 = STT_SEGMENT_MAX_INPUT_BYTES as u32; #[cfg(feature = "discord")] const DISCORD_TICK_SAMPLES: usize = (DISCORD_SAMPLE_RATE as usize / 50) * DISCORD_CHANNELS as usize; #[cfg(feature = "discord")] +const MAX_HOSTED_RECOVERY_WAV_SEGMENTS: u32 = 16; +#[cfg(feature = "discord")] +const MIN_HOSTED_RECOVERY_STALE_AFTER: Duration = Duration::from_secs(60); +#[cfg(feature = "discord")] +const HOSTED_START_STEP_TIMEOUT: Duration = Duration::from_secs(10); +#[cfg(feature = "discord")] +const HOSTED_AUTHORITY_WATCHDOG_INTERVAL: Duration = Duration::from_secs(1); +#[cfg(feature = "discord")] const DISCORD_PLAYOUT_BUFFER_PACKETS: u8 = 12; #[cfg(feature = "discord")] const DISCORD_PLAYOUT_SPIKE_PACKETS: u8 = 8; +#[cfg(feature = "discord")] +const DISCORD_VOICE_TRANSITION_TIMEOUT: Duration = Duration::from_secs(20); #[derive(Debug, Parser)] #[command(name = "call-scribe")] @@ -290,6 +307,52 @@ struct DiscordArgs { default_value = DEFAULT_ORGANIZATION_ID )] organization_id: String, + + /// Hosted control-plane origin. Requires --hosted-workload-token and disables + /// occupancy-based auto-start in favor of explicit durable commands. + #[arg( + long = "hosted-control-plane-url", + env = "CALL_SCRIBE_HOSTED_CONTROL_PLANE_URL" + )] + hosted_control_plane_url: Option, + + /// Dedicated workload credential used only for the hosted worker API. + #[arg( + long = "hosted-workload-token", + env = "CALL_SCRIBE_HOSTED_WORKLOAD_TOKEN" + )] + hosted_workload_token: Option, + + /// Stable secret used only to encrypt pending hosted usage leases at rest. + #[arg( + long = "hosted-outbox-encryption-key", + env = "CALL_SCRIBE_HOSTED_OUTBOX_ENCRYPTION_KEY" + )] + hosted_outbox_encryption_key: Option, + + /// Stable identifier included in hosted configuration and command requests. + #[arg( + long = "hosted-worker-id", + env = "CALL_SCRIBE_HOSTED_WORKER_ID", + default_value = "call-scribe-worker" + )] + hosted_worker_id: String, + + /// Hosted configuration and durable-command poll interval. + #[arg( + long = "hosted-poll-seconds", + env = "CALL_SCRIBE_HOSTED_POLL_SECONDS", + default_value_t = 15 + )] + hosted_poll_seconds: u64, + + /// Maximum configuration age. Recording fails closed after this interval. + #[arg( + long = "hosted-max-staleness-seconds", + env = "CALL_SCRIBE_HOSTED_MAX_STALENESS_SECONDS", + default_value_t = 60 + )] + hosted_max_staleness_seconds: u64, } #[cfg(feature = "discord")] @@ -327,6 +390,8 @@ pub(crate) async fn migrate_runtime_schema(pool: &PgPool) -> Result<()> { include_str!("../migrations/20260803120000_multi_tenant_recordings_transcripts.sql"), include_str!("../migrations/20260803140000_github_connections_and_issue_jobs.sql"), include_str!("../migrations/20260803160000_browser_sessions.sql"), + include_str!("../migrations/20260812010000_hosted_worker_command_executions.sql"), + include_str!("../migrations/20260812020000_hosted_capture_crash_recovery.sql"), ] { sqlx::raw_sql::raw_sql(migration) .execute(pool) @@ -424,16 +489,93 @@ struct DiscordCaptureConfig { allowed_channel_id: Option, runtime_store: Option, session_tx: mpsc::Sender, + hosted: Option, + self_hosted_organization_id: String, + self_hosted_capture_mode: CaptureMode, +} + +#[cfg(feature = "discord")] +#[derive(Clone)] +struct HostedCaptureConfig { + client: HostedControlPlaneClient, + configurations: HostedConfigurationStore, + poll_interval: Duration, } #[cfg(feature = "discord")] #[derive(Clone)] struct DiscordCaptureHandler { config: DiscordCaptureConfig, + recovery_owner_id: String, voice_states: Arc>>, - active: Arc>>, - reconcile_gate: Arc>, + active: Arc>, + finalizing_recording_ids: Arc>, + reconcile_gates: Arc>>>, bot_user_id: Arc>>, + requested_channels: Arc>, + hosted_poller_started: Arc, +} + +#[cfg(feature = "discord")] +#[derive(Clone, Debug, PartialEq, Eq)] +struct RequestedCapture { + channel_id: ChannelId, + generation: u64, + command_id: String, +} + +#[cfg(feature = "discord")] +enum DurableCommandClaim { + Claimed, + Indeterminate, + Completed { success: bool, result: Value }, +} + +#[cfg(feature = "discord")] +type HostedUsageOutboxRow = ( + String, + Vec, + Vec, + String, + i64, + DateTime, + DateTime, +); + +#[cfg(feature = "discord")] +type HostedCaptureRecoveryRow = ( + String, + Vec, + Vec, + String, + String, + i64, + DateTime, + DateTime, +); + +#[cfg(feature = "discord")] +struct FinalizingRecordingGuard { + finalizing: Arc>, + recording_id: String, +} + +#[cfg(feature = "discord")] +impl FinalizingRecordingGuard { + fn new(finalizing: Arc>, recording_id: String) -> Self { + finalizing.insert(recording_id.clone(), ()); + Self { + finalizing, + recording_id, + } + } +} + +#[cfg(feature = "discord")] +impl Drop for FinalizingRecordingGuard { + fn drop(&mut self) { + self.finalizing.remove(&self.recording_id); + } } #[cfg(feature = "discord")] @@ -452,9 +594,14 @@ struct ActiveCapture { guild_id: GuildId, channel_id: ChannelId, started_at: DateTime, + base_wav_path: PathBuf, recorder: SharedWavRecorder, known_ssrcs: Arc>, voice_stats: Arc, + runtime_store: Option, + capture_mode: CaptureMode, + hosted_usage: Option<(HostedControlPlaneClient, UsageReservation)>, + hosted_generation: Option, } #[cfg(feature = "discord")] @@ -466,6 +613,8 @@ struct CapturedSession { started_at: DateTime, stopped_at: DateTime, wav_paths: Vec, + runtime_store: Option, + capture_mode: CaptureMode, } #[cfg(feature = "discord")] @@ -514,6 +663,16 @@ impl SqlxRuntimeStore { migrate_runtime_schema(&self.pool).await } + async fn scoped(&self, organization_id: String, capture_mode: CaptureMode) -> Result { + let store = Self { + pool: self.pool.clone(), + organization_id, + capture_mode, + }; + store.ensure_organization().await?; + Ok(store) + } + async fn ensure_organization(&self) -> Result<()> { let name = if self.organization_id == DEFAULT_ORGANIZATION_ID { "Hazy Forge Private Alpha" @@ -844,153 +1003,808 @@ VALUES .with_context(|| format!("failed to record runtime audit event {}", event.event_type))?; Ok(()) } -} - -#[cfg(feature = "discord")] -type SharedWavRecorder = Arc>>; - -#[cfg(feature = "discord")] -struct SegmentedWavRecorder { - base_path: PathBuf, - segment_index: u32, - paths: Vec, - writer: hound::WavWriter>, - data_bytes_written: u32, - source_paths: HashMap, - source_writers: HashMap>>, -} - -#[cfg(feature = "discord")] -#[derive(Clone)] -struct DiscordVoiceReceiver { - guild_id: GuildId, - channel_id: ChannelId, - manager: Arc, - voice_states: Arc>>, - known_ssrcs: Arc>, - recorder: SharedWavRecorder, - ticks_since_flush: Arc, - voice_stats: Arc, - driver_refresh_gate: Arc>>, -} - -#[cfg(feature = "discord")] -#[derive(Default)] -struct DiscordVoiceStats { - speech_ticks: AtomicU64, - silent_ticks: AtomicU64, - decoded_ticks_by_ssrc: DashMap, - decoded_samples_by_ssrc: DashMap, -} -#[tokio::main] -async fn main() -> Result<()> { - let cli = Cli::parse(); + async fn claim_hosted_command(&self, command: &WorkerCommand) -> Result { + let generation = i64::try_from(command.generation) + .context("hosted command generation exceeded database range")?; + let inserted = sqlx::query::query( + r#" +INSERT INTO call_scribe_hosted_command_executions + (command_id, command_kind, guild_id, channel_id, generation, recording_notice_id, status) +VALUES ($1, $2, $3, $4, $5, $6, 'started') +ON CONFLICT (command_id) DO NOTHING +"#, + ) + .bind(&command.id) + .bind(&command.command_kind) + .bind(&command.guild_id) + .bind(&command.channel_id) + .bind(generation) + .bind(&command.recording_notice_id) + .execute(&self.pool) + .await + .context("failed to claim hosted command")?; + if inserted.rows_affected() == 1 { + return Ok(DurableCommandClaim::Claimed); + } - match cli.command { - Commands::Ingest(args) => ingest(args).await, - #[cfg(feature = "discord")] - Commands::Discord(args) => run_discord(args).await, - #[cfg(feature = "discord")] - Commands::RuntimeDb(args) => migrate_runtime_db(args).await, - Commands::Serve(args) => { - api::run_serve(api::ServeConfig { - database_url: args.database_url, - bind: args.bind, - meetings_dir: args.meetings_dir, - web_dir: args.web_dir, - stt_provider: args.provider, - organization_id: args.organization_id, - dev_auth_sub: args.dev_auth_sub, - oidc_issuer: args.oidc_issuer, - oidc_audience: args.oidc_audience, - oidc_client_id: args.oidc_client_id, - oidc_client_secret: args.oidc_client_secret, - public_origin: args.public_origin, - cookie_secure: args.cookie_secure, - github_token: args.github_token, - }) - .await + let row: ( + String, + String, + Option, + i64, + Option, + String, + Option, + ) = sqlx::query_as::query_as( + r#" +SELECT command_kind, guild_id, channel_id, generation, recording_notice_id, status, result +FROM call_scribe_hosted_command_executions +WHERE command_id = $1 +"#, + ) + .bind(&command.id) + .fetch_one(&self.pool) + .await + .context("failed to read hosted command execution")?; + if row.0 != command.command_kind + || row.1 != command.guild_id + || row.2 != command.channel_id + || row.3 != generation + || row.4 != command.recording_notice_id + { + bail!("command id was reused with different content"); + } + match row.5.as_str() { + "succeeded" => Ok(DurableCommandClaim::Completed { + success: true, + result: row.6.unwrap_or_else(|| serde_json::json!({})), + }), + "failed" => Ok(DurableCommandClaim::Completed { + success: false, + result: row.6.unwrap_or_else(|| { + serde_json::json!({ + "code": "command_rejected", + "message": "previous execution failed", + }) + }), + }), + _ => Ok(DurableCommandClaim::Indeterminate), } - #[cfg(not(feature = "discord"))] - Commands::Discord(_) => Err(anyhow::anyhow!( - "Discord capture was not compiled into this binary. Rebuild with `cargo run --features discord -- discord ...` after installing Opus/CMake dependencies." - )), } -} - -#[cfg(feature = "discord")] -async fn migrate_runtime_db(args: RuntimeDbArgs) -> Result<()> { - SqlxRuntimeStore::connect( - &args.database_url, - DEFAULT_ORGANIZATION_ID.to_string(), - CaptureMode::RecordOnly, - ) - .await?; - println!("Call Scribe runtime database is ready."); - Ok(()) -} -async fn ingest(args: IngestArgs) -> Result<()> { - let inputs = args - .input - .iter() - .map(|input| { - input - .canonicalize() - .with_context(|| format!("input recording does not exist: {}", input.display())) - }) - .collect::>>()?; - let input = inputs - .first() - .context("at least one input recording is required")? - .clone(); - let repo = args - .repo - .canonicalize() - .with_context(|| format!("target repo does not exist: {}", args.repo.display()))?; - ensure_repo_like(&repo)?; + async fn finish_hosted_command( + &self, + command_id: &str, + success: bool, + result: &Value, + ) -> Result<()> { + let status = if success { "succeeded" } else { "failed" }; + sqlx::query::query( + r#" +UPDATE call_scribe_hosted_command_executions +SET status = $2, result = $3, completed_at = now(), updated_at = now() +WHERE command_id = $1 AND status = 'started' +"#, + ) + .bind(command_id) + .bind(status) + .bind(result) + .execute(&self.pool) + .await + .context("failed to persist hosted command result")?; + Ok(()) + } - let title = args.title.unwrap_or_else(|| title_from_input(&input)); - if args.apply_docs { - let paths = transcribe_and_apply(ApplyRequest { - inputs, - repo, - title, - provider: args.provider, - language: args.language, - prompt: args.prompt, - output_dir: args.output_dir, - skip_analysis: args.skip_analysis, - }) - .await?; + async fn persist_hosted_capture_recovery( + &self, + client: &HostedControlPlaneClient, + reservation: &UsageReservation, + recording_id: &str, + base_wav_path: &Path, + started_at: DateTime, + owner_instance_id: &str, + ) -> Result<()> { + let base_wav_path = base_wav_path + .to_str() + .context("hosted capture path was not valid UTF-8")?; + let reserved_seconds = i64::try_from(reservation.reserved_seconds) + .context("hosted reservation duration exceeded database range")?; + let started_at = started_at.with_timezone(&Utc); + let expires_at = DateTime::parse_from_rfc3339(&reservation.expires_at) + .context("hosted reservation expiry was invalid")? + .with_timezone(&Utc); + if expires_at <= started_at { + bail!("hosted reservation expired before capture recovery could be persisted"); + } + let (encrypted_lease_token, encryption_nonce) = client + .encrypt_reservation_lease(&reservation.reservation_id, &reservation.lease_token)?; + sqlx::query::query( + r#" +INSERT INTO call_scribe_hosted_capture_recovery + (reservation_id, encrypted_lease_token, encryption_nonce, recording_id, + base_wav_path, reserved_seconds, started_at, expires_at, owner_instance_id) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) +"#, + ) + .bind(&reservation.reservation_id) + .bind(encrypted_lease_token) + .bind(encryption_nonce) + .bind(recording_id) + .bind(base_wav_path) + .bind(reserved_seconds) + .bind(started_at) + .bind(expires_at) + .bind(owner_instance_id) + .execute(&self.pool) + .await + .context("failed to persist hosted capture recovery state")?; + Ok(()) + } - println!("Wrote transcript package:"); - println!(" {}", paths.meeting_dir.display()); - println!("Transcript:"); - println!(" {}", paths.transcript.display()); - println!("Codex task:"); - println!(" {}", paths.codex_task.display()); - } else { - let rendered_transcript = - transcribe_recordings(&args.provider, &inputs, args.language, args.prompt).await?; - let path = write_standalone_markdown( - &repo.join(&args.output_dir), - &title, - Some(&input), - &rendered_transcript, + async fn remove_live_hosted_capture_recovery( + &self, + reservation_id: &str, + recording_id: &str, + owner_instance_id: &str, + ) -> Result<()> { + let deleted = sqlx::query::query( + r#" +DELETE FROM call_scribe_hosted_capture_recovery +WHERE reservation_id = $1 + AND recording_id = $2 + AND owner_instance_id = $3 + AND status = 'active' +"#, ) - .await?; - println!("Wrote Markdown transcript:"); - println!(" {}", path.display()); + .bind(reservation_id) + .bind(recording_id) + .bind(owner_instance_id) + .execute(&self.pool) + .await + .context("failed to clear hosted capture recovery state")?; + if deleted.rows_affected() != 1 { + bail!("live hosted capture recovery ownership was lost"); + } + Ok(()) } - Ok(()) -} + async fn claim_live_hosted_capture_finalization( + &self, + reservation_id: &str, + recording_id: &str, + owner_instance_id: &str, + authorization_ended_at: DateTime, + ) -> Result<(String, DateTime)> { + let recovery_claim_token = Uuid::new_v4().to_string(); + let effective_authorization_end: Option> = sqlx::query_scalar::query_scalar( + r#" +UPDATE call_scribe_hosted_capture_recovery +SET status = 'reconciling', + recovery_claim_token = $4, + recovery_lease_until = now() + interval '60 seconds', + expires_at = LEAST(expires_at, $5), + updated_at = now() +WHERE reservation_id = $1 + AND recording_id = $2 + AND owner_instance_id = $3 + AND status = 'active' + AND $5 > started_at +RETURNING expires_at +"#, + ) + .bind(reservation_id) + .bind(recording_id) + .bind(owner_instance_id) + .bind(&recovery_claim_token) + .bind(authorization_ended_at) + .fetch_optional(&self.pool) + .await + .context("failed to fence live hosted capture finalization")?; + let effective_authorization_end = effective_authorization_end + .context("live hosted capture recovery ownership was lost before finalization")?; + Ok((recovery_claim_token, effective_authorization_end)) + } -struct ApplyRequest { - inputs: Vec, - repo: PathBuf, + async fn remove_claimed_hosted_capture_recovery( + &self, + reservation_id: &str, + recording_id: &str, + recovery_claim_token: &str, + ) -> Result<()> { + let deleted = sqlx::query::query( + r#" +DELETE FROM call_scribe_hosted_capture_recovery +WHERE reservation_id = $1 + AND recording_id = $2 + AND status = 'reconciling' + AND recovery_claim_token = $3 +"#, + ) + .bind(reservation_id) + .bind(recording_id) + .bind(recovery_claim_token) + .execute(&self.pool) + .await + .context("failed to clear claimed hosted capture recovery state")?; + if deleted.rows_affected() != 1 { + bail!("hosted capture recovery claim fence was lost"); + } + Ok(()) + } + + async fn renew_hosted_capture_recovery( + &self, + owner_instance_id: &str, + reservation_id: &str, + recording_id: &str, + authorization_expires_at: DateTime, + ) -> Result<()> { + let minimum_expiry = Utc::now() + + chrono::Duration::from_std(RESERVATION_EXPIRY_MARGIN) + .expect("reservation expiry margin must fit chrono duration"); + if authorization_expires_at <= minimum_expiry { + bail!("hosted capture authority expires too soon to persist"); + } + let renewed = sqlx::query::query( + r#" +UPDATE call_scribe_hosted_capture_recovery +SET heartbeat_at = now(), expires_at = $4, updated_at = now() +WHERE owner_instance_id = $1 + AND reservation_id = $2 + AND recording_id = $3 + AND status = 'active' +"#, + ) + .bind(owner_instance_id) + .bind(reservation_id) + .bind(recording_id) + .bind(authorization_expires_at) + .execute(&self.pool) + .await + .context("failed to heartbeat live hosted capture recovery state")?; + if renewed.rows_affected() != 1 { + bail!("live hosted capture recovery ownership was lost"); + } + Ok(()) + } + + async fn defer_hosted_capture_recovery( + &self, + reservation_id: &str, + recovery_claim_token: &str, + error: &anyhow::Error, + ) -> Result<()> { + let message: String = format!("{error:#}").chars().take(500).collect(); + let deferred = sqlx::query::query( + r#" +UPDATE call_scribe_hosted_capture_recovery +SET status = 'active', + next_attempt_at = LEAST( + expires_at + interval '30 minutes' - interval '5 seconds', + now() + interval '30 seconds' + ), + recovery_lease_until = NULL, + recovery_claim_token = NULL, + last_error = $3, + updated_at = now() +WHERE reservation_id = $1 + AND status = 'reconciling' + AND recovery_claim_token = $2 +"#, + ) + .bind(reservation_id) + .bind(recovery_claim_token) + .bind(message) + .execute(&self.pool) + .await + .context("failed to defer hosted capture recovery")?; + if deferred.rows_affected() != 1 { + bail!("hosted capture recovery claim fence was lost before retry"); + } + Ok(()) + } + + async fn claim_abandoned_hosted_capture_recoveries( + &self, + stale_after: Duration, + recovery_claim_token: &str, + ) -> Result> { + let stale_after = chrono::Duration::from_std(stale_after) + .context("hosted recovery stale threshold exceeded timestamp range")?; + let stale_before = Utc::now() - stale_after; + sqlx::query_as::query_as( + r#" +WITH candidates AS ( + SELECT reservation_id + FROM call_scribe_hosted_capture_recovery + WHERE ((status = 'active' AND heartbeat_at <= $1) + OR (status = 'reconciling' AND recovery_lease_until <= now())) + AND next_attempt_at <= now() + AND expires_at + interval '30 minutes' > now() + interval '5 seconds' + AND encrypted_lease_token IS NOT NULL + AND encryption_nonce IS NOT NULL + ORDER BY next_attempt_at, created_at + FOR UPDATE SKIP LOCKED + LIMIT 25 +) +UPDATE call_scribe_hosted_capture_recovery AS recovery +SET status = 'reconciling', + recovery_claim_token = $2, + attempt_count = attempt_count + 1, + recovery_lease_until = now() + interval '60 seconds', + updated_at = now() +FROM candidates +WHERE recovery.reservation_id = candidates.reservation_id +RETURNING recovery.reservation_id, recovery.encrypted_lease_token, + recovery.encryption_nonce, recovery.recording_id, + recovery.base_wav_path, recovery.reserved_seconds, + recovery.started_at, recovery.expires_at +"#, + ) + .bind(stale_before) + .bind(recovery_claim_token) + .fetch_all(&self.pool) + .await + .context("failed to claim hosted capture recovery") + } + + async fn retry_hosted_capture_recovery( + &self, + client: &HostedControlPlaneClient, + capture_dir: &Path, + stale_after: Duration, + ) -> Result<()> { + let expired: Vec<(String, String)> = sqlx::query_as::query_as( + r#" +UPDATE call_scribe_hosted_capture_recovery +SET status = 'expired', + encrypted_lease_token = NULL, + encryption_nonce = NULL, + recovery_lease_until = NULL, + recovery_claim_token = NULL, + last_error = 'reservation expired before crash recovery completed', + updated_at = now() +WHERE (status = 'active' + OR (status = 'reconciling' AND recovery_lease_until <= now())) + AND expires_at + interval '30 minutes' <= now() + interval '5 seconds' +RETURNING reservation_id, recording_id +"#, + ) + .fetch_all(&self.pool) + .await + .context("failed to expire hosted capture recovery")?; + for (reservation_id, recording_id) in expired { + eprintln!( + "hosted capture {recording_id} reservation {reservation_id} expired before audio-derived recovery; operator reconciliation is required" + ); + } + + let recovery_claim_token = Uuid::new_v4().to_string(); + let recoveries = self + .claim_abandoned_hosted_capture_recoveries(stale_after, &recovery_claim_token) + .await?; + + for ( + reservation_id, + encrypted_lease_token, + encryption_nonce, + recording_id, + base_wav_path, + reserved_seconds, + started_at, + expires_at, + ) in recoveries + { + let recovery_result = async { + let base_wav_path = PathBuf::from(base_wav_path); + if !base_wav_path.starts_with(capture_dir) { + bail!("hosted recovery WAV escaped the configured capture directory"); + } + let lease_token = client.decrypt_reservation_lease( + &reservation_id, + &encrypted_lease_token, + &encryption_nonce, + )?; + let reserved_seconds = u64::try_from(reserved_seconds) + .context("hosted recovery reservation duration was invalid")?; + let reservation = UsageReservation { + reservation_id: reservation_id.clone(), + lease_token, + reserved_seconds, + expires_at: expires_at.to_rfc3339(), + }; + let actual_seconds = + recovered_usage_seconds(&base_wav_path, reserved_seconds)?.min( + authorized_usage_seconds(started_at, expires_at, reserved_seconds), + ); + if actual_seconds == 0 { + client.release_usage(&reservation).await?; + } else { + let occurred_at = started_at + + chrono::Duration::seconds( + i64::try_from(actual_seconds) + .context("hosted recovery duration exceeded timestamp range")?, + ); + self.enqueue_hosted_usage( + client, + &reservation, + &recording_id, + actual_seconds, + occurred_at, + ) + .await?; + } + self.remove_claimed_hosted_capture_recovery( + &reservation_id, + &recording_id, + &recovery_claim_token, + ) + .await + } + .await; + + if let Err(err) = recovery_result { + eprintln!("hosted capture {recording_id} recovery remains pending: {err:#}"); + self.defer_hosted_capture_recovery(&reservation_id, &recovery_claim_token, &err) + .await?; + } + } + Ok(()) + } + + async fn enqueue_hosted_usage( + &self, + client: &HostedControlPlaneClient, + reservation: &UsageReservation, + recording_id: &str, + actual_seconds: u64, + occurred_at: DateTime, + ) -> Result<()> { + let actual_seconds = i64::try_from(actual_seconds) + .context("hosted usage duration exceeded database range")?; + let authorization_expires_at = DateTime::parse_from_rfc3339(&reservation.expires_at) + .context("hosted usage expiry was invalid")? + .with_timezone(&Utc); + if authorization_expires_at < occurred_at { + bail!("hosted usage exceeded its capture authority"); + } + let expires_at = authorization_expires_at + + chrono::Duration::from_std(RESERVATION_SETTLEMENT_GRACE) + .expect("reservation settlement grace must fit chrono duration"); + if expires_at <= Utc::now() { + bail!("hosted usage settlement window expired before it could be queued"); + } + let (encrypted_lease_token, encryption_nonce) = client + .encrypt_reservation_lease(&reservation.reservation_id, &reservation.lease_token)?; + let inserted = sqlx::query::query( + r#" +INSERT INTO call_scribe_hosted_usage_outbox + (reservation_id, encrypted_lease_token, encryption_nonce, recording_id, + actual_seconds, occurred_at, expires_at) +VALUES ($1, $2, $3, $4, $5, $6, $7) +ON CONFLICT (reservation_id) DO NOTHING +"#, + ) + .bind(&reservation.reservation_id) + .bind(encrypted_lease_token) + .bind(encryption_nonce) + .bind(recording_id) + .bind(actual_seconds) + .bind(occurred_at) + .bind(expires_at) + .execute(&self.pool) + .await + .context("failed to enqueue hosted usage reconciliation")?; + if inserted.rows_affected() == 1 { + return Ok(()); + } + + let existing: (String, i64, DateTime, DateTime) = sqlx::query_as::query_as( + r#" +SELECT recording_id, actual_seconds, occurred_at, expires_at +FROM call_scribe_hosted_usage_outbox +WHERE reservation_id = $1 +"#, + ) + .bind(&reservation.reservation_id) + .fetch_one(&self.pool) + .await + .context("failed to verify existing hosted usage reconciliation")?; + if existing + != ( + recording_id.to_string(), + actual_seconds, + occurred_at, + expires_at, + ) + { + bail!("hosted reservation id was reused with different usage"); + } + Ok(()) + } + + async fn retry_hosted_usage_outbox(&self, client: &HostedControlPlaneClient) -> Result<()> { + let expired: Vec = sqlx::query_scalar::query_scalar( + r#" +UPDATE call_scribe_hosted_usage_outbox +SET status = 'expired', + encrypted_lease_token = NULL, + encryption_nonce = NULL, + last_error = 'reservation expired before usage delivery', + updated_at = now() +WHERE status = 'pending' AND expires_at <= now() + interval '5 seconds' +RETURNING reservation_id +"#, + ) + .fetch_all(&self.pool) + .await + .context("failed to expire hosted usage reconciliation")?; + for reservation_id in expired { + eprintln!( + "hosted usage reservation {reservation_id} expired before delivery; operator reconciliation is required" + ); + } + + let pending: Vec = sqlx::query_as::query_as( + r#" +SELECT reservation_id, encrypted_lease_token, encryption_nonce, recording_id, + actual_seconds, occurred_at, expires_at +FROM call_scribe_hosted_usage_outbox +WHERE status = 'pending' + AND next_attempt_at <= now() + AND expires_at > now() + interval '5 seconds' + AND encrypted_lease_token IS NOT NULL + AND encryption_nonce IS NOT NULL +ORDER BY next_attempt_at, created_at +LIMIT 25 +"#, + ) + .fetch_all(&self.pool) + .await + .context("failed to claim hosted usage reconciliation")?; + + for ( + reservation_id, + encrypted_lease_token, + encryption_nonce, + recording_id, + actual_seconds, + occurred_at, + expires_at, + ) in pending + { + let lease_token = match client.decrypt_reservation_lease( + &reservation_id, + &encrypted_lease_token, + &encryption_nonce, + ) { + Ok(lease_token) => lease_token, + Err(err) => { + let message: String = format!("{err:#}").chars().take(500).collect(); + sqlx::query::query( + r#" +UPDATE call_scribe_hosted_usage_outbox +SET status = 'expired', + encrypted_lease_token = NULL, + encryption_nonce = NULL, + last_error = $2, + updated_at = now() +WHERE reservation_id = $1 AND status = 'pending' +"#, + ) + .bind(&reservation_id) + .bind(message) + .execute(&self.pool) + .await + .context("failed to quarantine undecryptable hosted usage")?; + eprintln!( + "hosted usage reservation {reservation_id} cannot be decrypted; operator reconciliation is required" + ); + continue; + } + }; + let actual_seconds = u64::try_from(actual_seconds) + .context("queued hosted usage duration was invalid")?; + let reservation = UsageReservation { + reservation_id: reservation_id.clone(), + lease_token, + reserved_seconds: actual_seconds, + expires_at: expires_at.to_rfc3339(), + }; + match client + .consume_usage( + &reservation, + &recording_id, + actual_seconds, + &occurred_at.to_rfc3339(), + ) + .await + { + Ok(()) => { + sqlx::query::query( + r#" +UPDATE call_scribe_hosted_usage_outbox +SET status = 'delivered', attempt_count = attempt_count + 1, + encrypted_lease_token = NULL, encryption_nonce = NULL, + last_error = NULL, delivered_at = now(), updated_at = now() +WHERE reservation_id = $1 AND status = 'pending' +"#, + ) + .bind(&reservation_id) + .execute(&self.pool) + .await + .context("failed to mark hosted usage delivered")?; + } + Err(err) => { + let message: String = format!("{err:#}").chars().take(500).collect(); + sqlx::query::query( + r#" +UPDATE call_scribe_hosted_usage_outbox +SET attempt_count = attempt_count + 1, + next_attempt_at = LEAST(expires_at - interval '5 seconds', now() + interval '30 seconds'), + last_error = $2, + updated_at = now() +WHERE reservation_id = $1 AND status = 'pending' +"#, + ) + .bind(&reservation_id) + .bind(message) + .execute(&self.pool) + .await + .context("failed to defer hosted usage reconciliation")?; + } + } + } + Ok(()) + } +} + +#[cfg(feature = "discord")] +type SharedWavRecorder = Arc>>; + +#[cfg(feature = "discord")] +struct SegmentedWavRecorder { + base_path: PathBuf, + segment_index: u32, + paths: Vec, + writer: hound::WavWriter>, + data_bytes_written: u32, + source_paths: HashMap, + source_writers: HashMap>>, +} + +#[cfg(feature = "discord")] +#[derive(Clone)] +struct DiscordVoiceReceiver { + guild_id: GuildId, + channel_id: ChannelId, + manager: Arc, + voice_states: Arc>>, + known_ssrcs: Arc>, + recorder: SharedWavRecorder, + ticks_since_flush: Arc, + voice_stats: Arc, + driver_refresh_gate: Arc>>, +} + +#[cfg(feature = "discord")] +#[derive(Default)] +struct DiscordVoiceStats { + speech_ticks: AtomicU64, + silent_ticks: AtomicU64, + decoded_ticks_by_ssrc: DashMap, + decoded_samples_by_ssrc: DashMap, +} + +#[tokio::main] +async fn main() -> Result<()> { + let cli = Cli::parse(); + + match cli.command { + Commands::Ingest(args) => ingest(args).await, + #[cfg(feature = "discord")] + Commands::Discord(args) => run_discord(args).await, + #[cfg(feature = "discord")] + Commands::RuntimeDb(args) => migrate_runtime_db(args).await, + Commands::Serve(args) => { + api::run_serve(api::ServeConfig { + database_url: args.database_url, + bind: args.bind, + meetings_dir: args.meetings_dir, + web_dir: args.web_dir, + stt_provider: args.provider, + organization_id: args.organization_id, + dev_auth_sub: args.dev_auth_sub, + oidc_issuer: args.oidc_issuer, + oidc_audience: args.oidc_audience, + oidc_client_id: args.oidc_client_id, + oidc_client_secret: args.oidc_client_secret, + public_origin: args.public_origin, + cookie_secure: args.cookie_secure, + github_token: args.github_token, + }) + .await + } + #[cfg(not(feature = "discord"))] + Commands::Discord(_) => Err(anyhow::anyhow!( + "Discord capture was not compiled into this binary. Rebuild with `cargo run --features discord -- discord ...` after installing Opus/CMake dependencies." + )), + } +} + +#[cfg(feature = "discord")] +async fn migrate_runtime_db(args: RuntimeDbArgs) -> Result<()> { + SqlxRuntimeStore::connect( + &args.database_url, + DEFAULT_ORGANIZATION_ID.to_string(), + CaptureMode::RecordOnly, + ) + .await?; + println!("Call Scribe runtime database is ready."); + Ok(()) +} + +async fn ingest(args: IngestArgs) -> Result<()> { + let inputs = args + .input + .iter() + .map(|input| { + input + .canonicalize() + .with_context(|| format!("input recording does not exist: {}", input.display())) + }) + .collect::>>()?; + let input = inputs + .first() + .context("at least one input recording is required")? + .clone(); + let repo = args + .repo + .canonicalize() + .with_context(|| format!("target repo does not exist: {}", args.repo.display()))?; + ensure_repo_like(&repo)?; + + let title = args.title.unwrap_or_else(|| title_from_input(&input)); + if args.apply_docs { + let paths = transcribe_and_apply(ApplyRequest { + inputs, + repo, + title, + provider: args.provider, + language: args.language, + prompt: args.prompt, + output_dir: args.output_dir, + skip_analysis: args.skip_analysis, + }) + .await?; + + println!("Wrote transcript package:"); + println!(" {}", paths.meeting_dir.display()); + println!("Transcript:"); + println!(" {}", paths.transcript.display()); + println!("Codex task:"); + println!(" {}", paths.codex_task.display()); + } else { + let rendered_transcript = + transcribe_recordings(&args.provider, &inputs, args.language, args.prompt).await?; + let path = write_standalone_markdown( + &repo.join(&args.output_dir), + &title, + Some(&input), + &rendered_transcript, + ) + .await?; + println!("Wrote Markdown transcript:"); + println!(" {}", path.display()); + } + + Ok(()) +} + +struct ApplyRequest { + inputs: Vec, + repo: PathBuf, title: String, provider: SttProvider, language: Option, @@ -1047,6 +1861,49 @@ async fn transcribe_and_apply(request: ApplyRequest) -> Result { async fn run_discord(args: DiscordArgs) -> Result<()> { install_rustls_crypto_provider(); + let hosted = match ( + args.hosted_control_plane_url.as_deref(), + args.hosted_workload_token.clone(), + args.hosted_outbox_encryption_key.clone(), + ) { + (Some(base_url), Some(workload_token), Some(outbox_encryption_key)) => { + if args.hosted_poll_seconds == 0 { + bail!("hosted poll interval must be greater than zero"); + } + if args.hosted_max_staleness_seconds < args.hosted_poll_seconds { + bail!("hosted max staleness must be at least the hosted poll interval"); + } + if args.guild_id.is_some() || args.channel_id.is_some() || args.user_id.is_some() { + bail!( + "hosted mode cannot be combined with static guild, channel, or trigger-user configuration" + ); + } + if args.database_url.is_none() { + bail!( + "hosted mode requires CALL_SCRIBE_DATABASE_URL for durable command idempotency and per-guild session records" + ); + } + Some(HostedCaptureConfig { + client: HostedControlPlaneClient::new( + base_url, + workload_token, + args.hosted_worker_id.clone(), + outbox_encryption_key, + )?, + configurations: HostedConfigurationStore::new(Duration::from_secs( + args.hosted_max_staleness_seconds, + )), + poll_interval: Duration::from_secs(args.hosted_poll_seconds), + }) + } + (None, None, None) => None, + _ => { + bail!( + "hosted control-plane URL, workload token, and outbox encryption key must be configured together" + ) + } + }; + fs::create_dir_all(&args.capture_dir) .await .with_context(|| format!("failed to create {}", args.capture_dir.display()))?; @@ -1088,6 +1945,9 @@ async fn run_discord(args: DiscordArgs) -> Result<()> { allowed_channel_id: args.channel_id.map(ChannelId::new), runtime_store: runtime_store.clone(), session_tx, + hosted, + self_hosted_organization_id: args.organization_id.clone(), + self_hosted_capture_mode: args.capture_mode, }); let intents = GatewayIntents::GUILD_VOICE_STATES | GatewayIntents::GUILDS; @@ -1121,7 +1981,7 @@ async fn run_discord(args: DiscordArgs) -> Result<()> { result = &mut shutdown => { result.context("failed to listen for shutdown signal")?; println!("Stopping Discord capture after shutdown signal."); - if let Some(session) = handler.finalize_active_capture().await? { + for session in handler.finalize_all_active_captures().await? { println!( "Finalized active capture {} before shutdown; raw audio was retained for later processing.", session.id @@ -1138,8 +1998,6 @@ async fn run_discord(args: DiscordArgs) -> Result<()> { &args.output_dir, args.skip_analysis, args.apply_docs, - args.capture_mode, - runtime_store.as_ref(), ).await { eprintln!("post-capture processing failed; raw audio was retained: {err:#}"); } @@ -1192,9 +2050,9 @@ async fn handle_captured_session( output_dir: &Path, skip_analysis: bool, apply_docs: bool, - capture_mode: CaptureMode, - runtime_store: Option<&SqlxRuntimeStore>, ) -> Result<()> { + let capture_mode = session.capture_mode; + let runtime_store = session.runtime_store.as_ref(); let primary_wav_path = session .wav_paths .first() @@ -1397,10 +2255,14 @@ impl DiscordCaptureHandler { fn new(config: DiscordCaptureConfig) -> Self { Self { config, + recovery_owner_id: Uuid::new_v4().to_string(), voice_states: Arc::new(DashMap::new()), - active: Arc::new(Mutex::new(None)), - reconcile_gate: Arc::new(AsyncMutex::new(())), + active: Arc::new(DashMap::new()), + finalizing_recording_ids: Arc::new(DashMap::new()), + reconcile_gates: Arc::new(DashMap::new()), bot_user_id: Arc::new(Mutex::new(None)), + requested_channels: Arc::new(DashMap::new()), + hosted_poller_started: Arc::new(AtomicBool::new(false)), } } @@ -1416,27 +2278,65 @@ impl DiscordCaptureHandler { // Serenity may deliver voice-state updates while Songbird is joining or // leaving. Serialize those transitions so the next queued update always // evaluates the newest presence map after the prior I/O completes. - let _reconcile_guard = self.reconcile_gate.lock().await; + let reconcile_gate = self.reconcile_gate_for(guild_id); + let _reconcile_guard = reconcile_gate.lock().await; + + if let Some(hosted) = &self.config.hosted + && let Some(requested) = self + .requested_channels + .get(&guild_id) + .map(|requested| requested.clone()) + { + let still_current = hosted + .configurations + .policy_for(guild_id.get()) + .is_some_and(|policy| { + policy.desired_recording_generation == requested.generation + && policy.entitlement_active + && policy.recording_enabled + }); + if !still_current { + self.requested_channels.remove(&guild_id); + } + } let desired_channel = self.desired_capture_channel(guild_id); - let active_channel = { - self.active - .lock() - .expect("active capture mutex poisoned") - .as_ref() - .filter(|active| active.guild_id == guild_id) - .map(|active| active.channel_id) - }; + let active_channel = self.active.get(&guild_id).map(|active| active.channel_id); + let active_generation = self + .active + .get(&guild_id) + .and_then(|active| active.hosted_generation); + let requested_generation = self + .requested_channels + .get(&guild_id) + .map(|requested| requested.generation); + + let mut transition = capture_transition(active_channel, desired_channel); + if transition == CaptureTransition::Keep + && self.config.hosted.is_some() + && active_channel.is_some() + && active_generation != requested_generation + { + transition = CaptureTransition::Restart( + desired_channel.expect("hosted generation restart requires a desired channel"), + ); + } - match capture_transition(active_channel, desired_channel) { + match transition { CaptureTransition::Keep => {} CaptureTransition::Start(channel_id) => { if let Err(err) = self.start_capture(ctx, guild_id, channel_id).await { + if self.config.hosted.is_some() { + self.requested_channels.remove(&guild_id); + } eprintln!("failed to start Discord capture: {err:#}"); } } CaptureTransition::Stop => { + if self.config.hosted.is_some() { + self.requested_channels.remove(&guild_id); + } if let Err(err) = self.stop_capture(ctx, guild_id).await { eprintln!("failed to stop Discord capture: {err:#}"); } @@ -1447,6 +2347,9 @@ impl DiscordCaptureHandler { return; } if let Err(err) = self.start_capture(ctx, guild_id, channel_id).await { + if self.config.hosted.is_some() { + self.requested_channels.remove(&guild_id); + } eprintln!("failed to start Discord capture: {err:#}"); } } @@ -1454,6 +2357,29 @@ impl DiscordCaptureHandler { } fn desired_capture_channel(&self, guild_id: GuildId) -> Option { + if let Some(hosted) = &self.config.hosted { + let requested = self + .requested_channels + .get(&guild_id) + .map(|entry| entry.clone())?; + let policy = hosted.configurations.policy_for(guild_id.get())?; + let continuing_matching_reservation = + self.active.get(&guild_id).is_some_and(|active| { + active.channel_id == requested.channel_id + && active.hosted_generation == Some(requested.generation) + && active.hosted_usage.is_some() + }); + let policy_permits = if continuing_matching_reservation { + policy.permits_continuation(requested.channel_id.get()) + } else { + policy.permits_recording(requested.channel_id.get()) + }; + return (requested.generation == policy.desired_recording_generation + && policy_permits + && self.channel_has_users(guild_id, requested.channel_id)) + .then_some(requested.channel_id); + } + match (self.config.allowed_channel_id, self.config.trigger_user_id) { (Some(channel_id), Some(trigger_user_id)) if self.user_is_in_channel(guild_id, trigger_user_id, channel_id) => @@ -1495,6 +2421,48 @@ impl DiscordCaptureHandler { *self.bot_user_id.lock().expect("bot user id mutex poisoned") } + fn reconcile_gate_for(&self, guild_id: GuildId) -> Arc> { + self.reconcile_gates + .entry(guild_id) + .or_insert_with(|| Arc::new(AsyncMutex::new(()))) + .clone() + } + + async fn abort_starting_capture( + &self, + manager: &Arc, + guild_id: GuildId, + recorder: &SharedWavRecorder, + runtime_store: Option<&SqlxRuntimeStore>, + hosted_usage: Option<&(HostedControlPlaneClient, UsageReservation)>, + recover_owned_reservation: bool, + ) { + // No hosted failure cleanup may wait on Discord or the database while + // its writer remains enabled. Closing the WAV is the side-effect fence. + if let Err(err) = finalize_wav(recorder) { + eprintln!("failed to checkpoint aborted Discord capture: {err:#}"); + } + if manager.get(guild_id).is_some() + && tokio::time::timeout(DISCORD_VOICE_TRANSITION_TIMEOUT, manager.remove(guild_id)) + .await + .is_err() + { + eprintln!( + "timed out leaving Discord after aborting capture in guild {}", + guild_id.get() + ); + } + if recover_owned_reservation + && let Some((client, _)) = hosted_usage + && let Some(store) = runtime_store + && let Err(err) = store + .retry_hosted_capture_recovery(client, &self.config.capture_dir, Duration::ZERO) + .await + { + eprintln!("hosted usage recovery remains pending after aborted start: {err:#}"); + } + } + async fn start_capture( &self, ctx: &SerenityContext, @@ -1503,14 +2471,61 @@ impl DiscordCaptureHandler { ) -> Result<()> { self.stop_capture(ctx, guild_id).await?; + let (organization_id, capture_mode, hosted_generation, hosted_reservation_request) = + if let Some(hosted) = &self.config.hosted { + let policy = hosted + .configurations + .policy_for(guild_id.get()) + .context("hosted configuration is missing or stale")?; + if !policy.permits_recording(channel_id.get()) { + bail!("hosted configuration does not authorize this guild and channel"); + } + let requested = self + .requested_channels + .get(&guild_id) + .map(|requested| requested.clone()) + .context("hosted start has no durable requested state")?; + if requested.channel_id != channel_id + || requested.generation != policy.desired_recording_generation + { + bail!("hosted start generation does not match current desired state"); + } + let requested_seconds = policy + .remaining_recording_seconds + .context("hosted remaining usage is not configured")? + .min(3_600); + ( + policy.organization_id, + CaptureMode::RecordOnly, + Some(requested.generation), + Some(( + hosted.client.clone(), + requested.command_id, + requested_seconds, + )), + ) + } else { + ( + self.config.self_hosted_organization_id.clone(), + self.config.self_hosted_capture_mode, + None, + None, + ) + }; + let runtime_store = match &self.config.runtime_store { + Some(store) => Some(store.scoped(organization_id, capture_mode).await?), + None => None, + }; + let session_id = Uuid::new_v4().to_string(); let started_at = Local::now(); let title = format!("Discord call {}", started_at.format("%Y-%m-%d %H%M")); let base_wav_path = self.config.capture_dir.join(format!( - "{}-guild-{}-channel-{}.wav", + "{}-guild-{}-channel-{}-session-{}.wav", started_at.format("%Y%m%d-%H%M%S"), guild_id.get(), - channel_id.get() + channel_id.get(), + session_id, )); let recorder = create_wav_recorder(&base_wav_path)?; let known_ssrcs = Arc::new(DashMap::new()); @@ -1531,38 +2546,226 @@ impl DiscordCaptureHandler { driver_refresh_gate: Arc::new(Mutex::new(None)), }; - manager - .join(guild_id, channel_id) - .await - .map_err(|err| anyhow::anyhow!("failed to join Discord voice channel: {err:?}"))?; - - { - let handler_lock = manager.get_or_insert(guild_id); - let mut handler = handler_lock.lock().await; - handler.add_global_event(CoreEvent::SpeakingStateUpdate.into(), receiver.clone()); - handler.add_global_event(CoreEvent::ClientDisconnect.into(), receiver.clone()); - handler.add_global_event(CoreEvent::VoiceTick.into(), receiver); - } + // Reserve last, immediately before joining voice, so local setup + // failures cannot strand a control-plane quota lease. + let mut hosted_usage = match hosted_reservation_request { + Some((client, command_id, requested_seconds)) => { + let reservation = client + .reserve_usage(&command_id, requested_seconds) + .await + .context("failed to reserve hosted recording usage")?; + Some((client, reservation)) + } + None => None, + }; - if let Some(store) = &self.config.runtime_store - && let Err(err) = store - .record_session_started( + if let Some((client, reservation)) = &hosted_usage { + let store = runtime_store + .as_ref() + .context("hosted capture omitted its durable runtime store")?; + if let Err(err) = store + .persist_hosted_capture_recovery( + client, + reservation, &session_id, - guild_id, - channel_id, + &base_wav_path, started_at, - &title, - serde_json::json!({ - "base_wav_path": base_wav_path.display().to_string(), - "capture_dir": self.config.capture_dir.display().to_string(), - }), + &self.recovery_owner_id, ) .await - { - let _ = manager.remove(guild_id).await; + { + if client.release_usage(reservation).await.is_ok() { + let _ = store + .remove_live_hosted_capture_recovery( + &reservation.reservation_id, + &session_id, + &self.recovery_owner_id, + ) + .await; + } + return Err(err).context( + "hosted reservation was not durably recoverable before joining voice", + ); + } + } + + let join_result = tokio::time::timeout( + DISCORD_VOICE_TRANSITION_TIMEOUT, + manager.join(guild_id, channel_id), + ) + .await; + let join_error = match &join_result { + Ok(Ok(_)) => None, + Ok(Err(err)) => Some(format!("failed to join Discord voice channel: {err:?}")), + Err(_) => Some("timed out joining Discord voice channel".to_string()), + }; + if let Some(join_error) = join_error { + if let Some((client, reservation)) = &hosted_usage { + match client.release_usage(reservation).await { + Ok(()) => { + if let Some(store) = &runtime_store + && let Err(clear_err) = store + .remove_live_hosted_capture_recovery( + &reservation.reservation_id, + &session_id, + &self.recovery_owner_id, + ) + .await + { + eprintln!( + "failed to clear released hosted capture recovery: {clear_err:#}" + ); + } + } + Err(release_err) => { + eprintln!( + "failed to release hosted usage after voice join failure; recovery remains pending: {release_err:#}" + ); + } + } + } + bail!(join_error); + } + + let session_metadata = serde_json::json!({ + "base_wav_path": base_wav_path.display().to_string(), + "capture_dir": self.config.capture_dir.display().to_string(), + "hosted_generation": hosted_generation, + "usage_reservation_id": hosted_usage + .as_ref() + .map(|(_, reservation)| &reservation.reservation_id), + }); + let session_start_result = if let Some(store) = &runtime_store { + let persist = store.record_session_started( + &session_id, + guild_id, + channel_id, + started_at, + &title, + session_metadata, + ); + if hosted_usage.is_some() { + match tokio::time::timeout(HOSTED_START_STEP_TIMEOUT, persist).await { + Ok(result) => result, + Err(_) => Err(anyhow::anyhow!( + "timed out persisting hosted session start before ownership deadline" + )), + } + } else { + persist.await + } + } else { + Ok(()) + }; + if let Err(err) = session_start_result { + self.abort_starting_capture( + &manager, + guild_id, + &recorder, + runtime_store.as_ref(), + hosted_usage.as_ref(), + true, + ) + .await; return Err(err); } + let handler_lock = manager.get_or_insert(guild_id); + let mut handler = if hosted_usage.is_some() { + match tokio::time::timeout(HOSTED_START_STEP_TIMEOUT, handler_lock.lock()).await { + Ok(handler) => handler, + Err(_) => { + self.abort_starting_capture( + &manager, + guild_id, + &recorder, + runtime_store.as_ref(), + hosted_usage.as_ref(), + true, + ) + .await; + bail!("timed out preparing hosted voice handlers before ownership deadline"); + } + } + } else { + handler_lock.lock().await + }; + + let hosted_usage_for_abort = hosted_usage.clone(); + if let Some((client, reservation)) = hosted_usage.as_mut() { + let store = runtime_store + .as_ref() + .context("hosted capture omitted its durable runtime store")?; + let heartbeat = tokio::time::timeout( + HOSTED_START_STEP_TIMEOUT, + client.heartbeat_usage(reservation), + ) + .await; + let renewed_expires_at = match heartbeat { + Ok(Ok(expires_at)) => expires_at, + Ok(Err(err)) => { + drop(handler); + self.abort_starting_capture( + &manager, + guild_id, + &recorder, + runtime_store.as_ref(), + hosted_usage_for_abort.as_ref(), + true, + ) + .await; + return Err(err).context( + "hosted capture authority was not renewable before voice handlers enabled", + ); + } + Err(_) => { + drop(handler); + self.abort_starting_capture( + &manager, + guild_id, + &recorder, + runtime_store.as_ref(), + hosted_usage_for_abort.as_ref(), + true, + ) + .await; + bail!( + "hosted capture authority renewal timed out before voice handlers enabled" + ); + } + }; + reservation.expires_at = renewed_expires_at.to_rfc3339(); + let ownership = tokio::time::timeout( + HOSTED_START_STEP_TIMEOUT, + store.renew_hosted_capture_recovery( + &self.recovery_owner_id, + &reservation.reservation_id, + &session_id, + renewed_expires_at, + ), + ) + .await; + if !matches!(ownership, Ok(Ok(()))) { + drop(handler); + self.abort_starting_capture( + &manager, + guild_id, + &recorder, + runtime_store.as_ref(), + hosted_usage_for_abort.as_ref(), + false, + ) + .await; + bail!( + "hosted capture recovery ownership was not renewable before voice handlers enabled" + ); + } + } + handler.add_global_event(CoreEvent::SpeakingStateUpdate.into(), receiver.clone()); + handler.add_global_event(CoreEvent::ClientDisconnect.into(), receiver.clone()); + handler.add_global_event(CoreEvent::VoiceTick.into(), receiver); + drop(handler); + println!( "Started Discord capture in guild {} channel {} -> {}", guild_id.get(), @@ -1570,60 +2773,152 @@ impl DiscordCaptureHandler { base_wav_path.display() ); - let mut active = self.active.lock().expect("active capture mutex poisoned"); - *active = Some(ActiveCapture { - session_id, + let maximum_duration = hosted_usage + .as_ref() + .map(|(_, reservation)| Duration::from_secs(reservation.reserved_seconds)); + let expiry_session_id = session_id.clone(); + self.active.insert( guild_id, - channel_id, - started_at, - recorder, - known_ssrcs, - voice_stats, - }); + ActiveCapture { + session_id, + guild_id, + channel_id, + started_at, + base_wav_path, + recorder, + known_ssrcs, + voice_stats, + runtime_store, + capture_mode, + hosted_usage, + hosted_generation, + }, + ); + if let Some(maximum_duration) = maximum_duration { + let handler = self.clone(); + let ctx = ctx.clone(); + tokio::spawn(async move { + handler + .run_hosted_capture_watchdog(ctx, guild_id, expiry_session_id, maximum_duration) + .await; + }); + } Ok(()) } async fn stop_capture(&self, ctx: &SerenityContext, guild_id: GuildId) -> Result<()> { - let active_guild_id = self - .active - .lock() - .expect("active capture mutex poisoned") - .as_ref() - .map(|active| active.guild_id); - - if active_guild_id == Some(guild_id) { + if self.active.contains_key(&guild_id) { let manager = songbird::get(ctx) .await .context("Songbird voice client was not registered")? .clone(); if manager.get(guild_id).is_some() { - manager.remove(guild_id).await.map_err(|err| { - anyhow::anyhow!("failed to leave Discord voice channel: {err:?}") - })?; + tokio::time::timeout(DISCORD_VOICE_TRANSITION_TIMEOUT, manager.remove(guild_id)) + .await + .context("timed out leaving Discord voice channel")? + .map_err(|err| { + anyhow::anyhow!("failed to leave Discord voice channel: {err:?}") + })?; } } - if active_guild_id == Some(guild_id) - && let Some(session) = self.finalize_active_capture().await? - { + if let Some(session) = self.finalize_active_capture(guild_id).await? { let _ = self.config.session_tx.send(session).await; } Ok(()) } - async fn finalize_active_capture(&self) -> Result> { - let active = { - let mut active = self.active.lock().expect("active capture mutex poisoned"); - active.take() - }; - let Some(active) = active else { + async fn finalize_active_capture(&self, guild_id: GuildId) -> Result> { + let Some((_, active)) = self.active.remove(&guild_id) else { return Ok(None); }; - let wav_paths = finalize_wav(&active.recorder)?; - active.voice_stats.print(&active.known_ssrcs); + let _finalizing_guard = FinalizingRecordingGuard::new( + self.finalizing_recording_ids.clone(), + active.session_id.clone(), + ); + let result = self.finalize_removed_capture(active).await; + result.map(Some) + } + + async fn finalize_removed_capture(&self, active: ActiveCapture) -> Result { let stopped_at = Local::now(); - if let Some(store) = &self.config.runtime_store { + let mut wav_paths = finalize_wav(&active.recorder)?; + if wav_paths.is_empty() && active.hosted_usage.is_some() { + // A DB-heartbeat failure self-fences the writer before any network + // cleanup. Reconstruct the already-finalized contiguous path list + // so normal usage/session finalization can still proceed if the + // database becomes reachable during shutdown. + wav_paths = checkpointed_wav_paths(&active.base_wav_path)?; + } + active.voice_stats.print(&active.known_ssrcs); + if let Some((client, reservation)) = &active.hosted_usage { + let base_wav_path = wav_paths + .first() + .context("hosted capture did not retain its mixed-audio WAV")?; + let recovered_seconds = + recovered_usage_seconds(base_wav_path, reservation.reserved_seconds)?; + let store = active + .runtime_store + .as_ref() + .context("hosted capture omitted its durable runtime store")?; + let last_authorization_expiry = DateTime::parse_from_rfc3339(&reservation.expires_at) + .context("hosted reservation expiry was invalid during finalization")? + .with_timezone(&Utc); + let requested_authorization_end = + std::cmp::min(last_authorization_expiry, stopped_at.with_timezone(&Utc)); + let (recovery_claim_token, authorization_ended_at) = store + .claim_live_hosted_capture_finalization( + &reservation.reservation_id, + &active.session_id, + &self.recovery_owner_id, + requested_authorization_end, + ) + .await?; + let actual_seconds = recovered_seconds.min(authorized_usage_seconds( + active.started_at.with_timezone(&Utc), + authorization_ended_at, + reservation.reserved_seconds, + )); + let mut settlement_reservation = reservation.clone(); + settlement_reservation.expires_at = authorization_ended_at.to_rfc3339(); + if actual_seconds == 0 { + client + .release_usage(&settlement_reservation) + .await + .context("failed to release zero-duration hosted capture")?; + } else { + let occurred_at = active.started_at.with_timezone(&Utc) + + chrono::Duration::seconds( + i64::try_from(actual_seconds) + .context("hosted usage duration exceeded timestamp range")?, + ); + store + .enqueue_hosted_usage( + client, + &settlement_reservation, + &active.session_id, + actual_seconds, + occurred_at, + ) + .await + .context("failed to durably enqueue hosted usage")?; + } + store + .remove_claimed_hosted_capture_recovery( + &reservation.reservation_id, + &active.session_id, + &recovery_claim_token, + ) + .await?; + if let Err(err) = store.retry_hosted_usage_outbox(client).await { + eprintln!( + "hosted usage for recording {} remains queued for retry: {err:#}", + active.session_id + ); + } + } + if let Some(store) = &active.runtime_store { if let Err(err) = store .record_session_stopped(&active.session_id, stopped_at) .await @@ -1644,17 +2939,519 @@ impl DiscordCaptureHandler { } } } - Ok(Some(CapturedSession { + Ok(CapturedSession { id: active.session_id, guild_id: active.guild_id, channel_id: active.channel_id, started_at: active.started_at, stopped_at, wav_paths, - })) + runtime_store: active.runtime_store, + capture_mode: active.capture_mode, + }) + } + + async fn finalize_all_active_captures(&self) -> Result> { + let guild_ids: Vec<_> = self.active.iter().map(|entry| *entry.key()).collect(); + let mut sessions = Vec::new(); + for guild_id in guild_ids { + if let Some(session) = self.finalize_active_capture(guild_id).await? { + sessions.push(session); + } + } + Ok(sessions) + } + + fn self_fence_hosted_recorders(&self, guild_ids: &[GuildId]) { + for guild_id in guild_ids { + self.requested_channels.remove(guild_id); + let Some(mut active) = self.active.get_mut(guild_id) else { + continue; + }; + let Some((_, reservation)) = active.hosted_usage.as_mut() else { + continue; + }; + let fence_at = Utc::now(); + let authorization_end = DateTime::parse_from_rfc3339(&reservation.expires_at) + .map(|expires_at| std::cmp::min(expires_at.with_timezone(&Utc), fence_at)) + .unwrap_or(fence_at); + reservation.expires_at = authorization_end.to_rfc3339(); + if let Err(err) = finalize_wav(&active.recorder) { + eprintln!( + "failed to finalize self-fenced hosted recording {}: {err:#}", + active.session_id + ); + } + } + } + + async fn finish_self_fenced_hosted_capture(&self, ctx: &SerenityContext, guild_id: GuildId) { + let reconcile_gate = self.reconcile_gate_for(guild_id); + let _reconcile_guard = reconcile_gate.lock().await; + match songbird::get(ctx).await { + Some(manager) if manager.get(guild_id).is_some() => { + match tokio::time::timeout( + DISCORD_VOICE_TRANSITION_TIMEOUT, + manager.remove(guild_id), + ) + .await + { + Ok(Ok(())) => {} + Ok(Err(err)) => eprintln!( + "failed to leave Discord after hosted authority self-fence in guild {}: {err:?}", + guild_id.get() + ), + Err(_) => eprintln!( + "timed out leaving Discord after hosted authority self-fence in guild {}", + guild_id.get() + ), + } + } + _ => {} + } + match self.finalize_active_capture(guild_id).await { + Ok(Some(session)) => { + let _ = self.config.session_tx.send(session).await; + } + Ok(None) => {} + Err(err) => eprintln!( + "hosted capture in guild {} remains durably recoverable after self-fence: {err:#}", + guild_id.get() + ), + } + } + + async fn run_hosted_capture_watchdog( + self, + ctx: SerenityContext, + guild_id: GuildId, + session_id: String, + maximum_duration: Duration, + ) { + let usage_deadline = Instant::now() + maximum_duration; + let mut interval = tokio::time::interval(HOSTED_AUTHORITY_WATCHDOG_INTERVAL); + interval.set_missed_tick_behavior(MissedTickBehavior::Delay); + loop { + interval.tick().await; + let fence_reason = { + let Some(active) = self + .active + .get(&guild_id) + .filter(|active| active.session_id == session_id) + else { + return; + }; + let Some((_, reservation)) = &active.hosted_usage else { + return; + }; + if Instant::now() >= usage_deadline { + Some("reserved usage duration reached") + } else { + match DateTime::parse_from_rfc3339(&reservation.expires_at) { + Ok(expires_at) + if hosted_authority_requires_fence( + expires_at.with_timezone(&Utc), + Utc::now(), + ) => + { + Some("last renewed authority is near expiry") + } + Ok(_) => None, + Err(_) => Some("last renewed authority expiry is malformed"), + } + } + }; + let Some(fence_reason) = fence_reason else { + continue; + }; + eprintln!( + "hosted capture in guild {} is self-fencing because {fence_reason}", + guild_id.get() + ); + self.self_fence_hosted_recorders(&[guild_id]); + self.finish_self_fenced_hosted_capture(&ctx, guild_id).await; + return; + } + } + + async fn run_hosted_heartbeat_monitor( + self, + ctx: SerenityContext, + heartbeat_interval: Duration, + ) { + let mut interval = tokio::time::interval(heartbeat_interval); + interval.set_missed_tick_behavior(MissedTickBehavior::Delay); + loop { + interval.tick().await; + let Some(store) = &self.config.runtime_store else { + return; + }; + let active_hosted_captures: Vec<_> = self + .active + .iter() + .filter_map(|active| { + active.hosted_usage.as_ref().map(|(client, reservation)| { + ( + *active.key(), + active.session_id.clone(), + client.clone(), + reservation.clone(), + ) + }) + }) + .collect(); + if active_hosted_captures.is_empty() { + continue; + } + let expected: Vec<_> = active_hosted_captures + .iter() + .map(|(guild_id, recording_id, _, _)| (*guild_id, recording_id.clone())) + .collect(); + let mut heartbeats = tokio::task::JoinSet::new(); + for (guild_id, recording_id, client, reservation) in active_hosted_captures { + let store = store.clone(); + let owner_instance_id = self.recovery_owner_id.clone(); + heartbeats.spawn(async move { + let reservation_id = reservation.reservation_id.clone(); + let result = tokio::time::timeout(heartbeat_interval, async { + let expires_at = client.heartbeat_usage(&reservation).await?; + store + .renew_hosted_capture_recovery( + &owner_instance_id, + &reservation_id, + &recording_id, + expires_at, + ) + .await?; + Ok::<_, anyhow::Error>(expires_at) + }) + .await + .map_err(|_| anyhow::anyhow!("hosted usage heartbeat timed out")) + .and_then(|result| result); + (guild_id, recording_id, reservation_id, result) + }); + } + + let mut heartbeated = std::collections::HashSet::new(); + while let Some(result) = heartbeats.join_next().await { + match result { + Ok((guild_id, recording_id, reservation_id, Ok(expires_at))) => { + if let Some(mut active) = self.active.get_mut(&guild_id) + && active.session_id == recording_id + && let Some((_, reservation)) = active.hosted_usage.as_mut() + && reservation.reservation_id == reservation_id + { + reservation.expires_at = expires_at.to_rfc3339(); + } + heartbeated.insert(recording_id); + } + Ok((guild_id, recording_id, reservation_id, Err(err))) => { + eprintln!( + "hosted usage heartbeat failed for guild {} recording {} reservation {}; self-fencing capture: {err:#}", + guild_id.get(), + recording_id, + reservation_id + ); + } + Err(err) => { + eprintln!( + "hosted usage heartbeat task failed; affected capture will self-fence: {err}" + ); + } + } + } + let unowned_guilds = hosted_guilds_without_heartbeat(&expected, Some(&heartbeated)); + if unowned_guilds.is_empty() { + continue; + } + + // Close all affected writers synchronously before voice/network/DB + // cleanup. At a ten-second maximum monitor interval and a 60-second + // minimum stale threshold, the old owner stops before takeover. + self.self_fence_hosted_recorders(&unowned_guilds); + for guild_id in unowned_guilds { + self.finish_self_fenced_hosted_capture(&ctx, guild_id).await; + } + } + } + + async fn run_hosted_poller(self, ctx: SerenityContext, hosted: HostedCaptureConfig) { + let heartbeat_interval = hosted + .poll_interval + .min(Duration::from_secs(10)) + .max(Duration::from_secs(1)); + let heartbeat_handler = self.clone(); + let heartbeat_ctx = ctx.clone(); + tokio::spawn(async move { + heartbeat_handler + .run_hosted_heartbeat_monitor(heartbeat_ctx, heartbeat_interval) + .await; + }); + + let mut interval = tokio::time::interval(hosted.poll_interval); + interval.set_missed_tick_behavior(MissedTickBehavior::Delay); + loop { + interval.tick().await; + + if let Some(store) = &self.config.runtime_store { + let recovery_stale_after = heartbeat_interval + .saturating_mul(6) + .max(MIN_HOSTED_RECOVERY_STALE_AFTER); + if let Err(err) = store + .retry_hosted_capture_recovery( + &hosted.client, + &self.config.capture_dir, + recovery_stale_after, + ) + .await + { + eprintln!("hosted active-capture recovery failed: {err:#}"); + } + if let Err(err) = store.retry_hosted_usage_outbox(&hosted.client).await { + eprintln!("hosted usage outbox retry failed: {err:#}"); + } + } + + let mut reconcile_guilds = std::collections::HashSet::new(); + match hosted.client.fetch_configurations().await { + Ok(response) => { + let revision = response.revision.clone(); + reconcile_guilds.extend(hosted.configurations.replace(response)); + println!("Applied hosted worker configuration revision {revision}"); + } + Err(err) => { + eprintln!( + "hosted configuration refresh failed; stale or missing policies deny recording: {err:#}" + ); + reconcile_guilds.extend(hosted.configurations.guild_ids()); + } + } + reconcile_guilds.extend(self.active.iter().map(|entry| entry.key().get())); + reconcile_guilds.extend( + self.requested_channels + .iter() + .map(|entry| entry.key().get()), + ); + let mut reconciliations = tokio::task::JoinSet::new(); + for guild_id in reconcile_guilds { + let handler = self.clone(); + let ctx = ctx.clone(); + reconciliations.spawn(async move { + handler + .reconcile_capture(&ctx, GuildId::new(guild_id)) + .await; + }); + } + while let Some(result) = reconciliations.join_next().await { + if let Err(err) = result { + eprintln!("hosted guild reconciliation task failed: {err}"); + } + } + + match hosted.client.lease_commands().await { + Ok(response) => { + for command in response.commands { + let Some(store) = &self.config.runtime_store else { + eprintln!("hosted command skipped because runtime database is absent"); + continue; + }; + let (success, result, should_persist) = match store + .claim_hosted_command(&command) + .await + { + Ok(DurableCommandClaim::Completed { success, result }) => { + (success, result, false) + } + Ok(DurableCommandClaim::Indeterminate) => ( + false, + serde_json::json!({ + "code": "command_indeterminate", + "message": "a previous worker stopped during this command; no side effect was replayed", + }), + false, + ), + Ok(DurableCommandClaim::Claimed) => { + match self.execute_hosted_command(&ctx, &command).await { + Ok(result) => (true, result, true), + Err(err) => { + let message: String = + format!("{err:#}").chars().take(200).collect(); + ( + false, + serde_json::json!({ + "code": "command_rejected", + "message": message, + }), + true, + ) + } + } + } + Err(err) => { + eprintln!("failed to claim hosted command {}: {err:#}", command.id); + continue; + } + }; + if should_persist + && let Err(err) = store + .finish_hosted_command(&command.id, success, &result) + .await + { + eprintln!( + "failed to durably finish hosted command {}; acknowledgement withheld: {err:#}", + command.id + ); + continue; + } + if let Err(err) = hosted + .client + .complete_command(&command.id, &command.lease_token, success, result) + .await + { + eprintln!( + "failed to acknowledge hosted command {}: {err:#}", + command.id + ); + } + } + } + Err(err) => { + eprintln!("hosted durable-command poll failed: {err:#}"); + } + } + } + } + + async fn execute_hosted_command( + &self, + ctx: &SerenityContext, + command: &WorkerCommand, + ) -> Result { + let guild_id = command + .guild_id + .parse::() + .context("command guild id was invalid")?; + let guild_id = GuildId::new(guild_id); + match command.command_kind.as_str() { + "record_start" => { + if command + .recording_notice_id + .as_deref() + .is_none_or(|notice_id| notice_id.trim().is_empty()) + { + bail!("start command omitted durable recording notice evidence"); + } + let channel_id = command + .channel_id + .as_deref() + .context("start command omitted channel id")? + .parse::() + .context("command channel id was invalid")?; + let hosted = self + .config + .hosted + .as_ref() + .context("hosted command received outside hosted mode")?; + let policy = hosted + .configurations + .policy_for(guild_id.get()) + .context("hosted configuration is missing or stale")?; + if command.generation != policy.desired_recording_generation { + bail!("start command generation is stale or ahead of desired state"); + } + if !policy.permits_recording(channel_id) { + bail!("guild or channel is not authorized to record"); + } + let channel_id = ChannelId::new(channel_id); + if !self.channel_has_users(guild_id, channel_id) { + bail!("approved voice channel has no human participant"); + } + if self + .requested_channels + .get(&guild_id) + .is_some_and(|requested| requested.generation > command.generation) + { + bail!("start command generation is older than pending desired state"); + } + self.requested_channels.insert( + guild_id, + RequestedCapture { + channel_id, + generation: command.generation, + command_id: command.id.clone(), + }, + ); + self.reconcile_capture(ctx, guild_id).await; + if !self.active.get(&guild_id).is_some_and(|active| { + active.channel_id == channel_id + && active.hosted_generation == Some(command.generation) + }) { + if self + .requested_channels + .get(&guild_id) + .is_some_and(|requested| requested.generation == command.generation) + { + self.requested_channels.remove(&guild_id); + } + bail!("worker could not start the requested capture"); + } + let session_id = self + .active + .get(&guild_id) + .map(|active| active.session_id.clone()) + .context("started capture did not expose its recording id")?; + Ok(serde_json::json!({ + "code": "recording_started", + "recordingId": session_id, + })) + } + "record_stop" => { + let reconcile_gate = self.reconcile_gate_for(guild_id); + let _reconcile_guard = reconcile_gate.lock().await; + if !self.stop_generation_is_not_older(guild_id, command.generation) { + bail!("stop command generation is older than current desired state"); + } + self.requested_channels.remove(&guild_id); + let should_stop = self.active.contains_key(&guild_id); + if should_stop { + self.stop_capture(ctx, guild_id).await?; + } + if self.active.contains_key(&guild_id) { + bail!("worker could not stop the requested capture"); + } + Ok(serde_json::json!({ "code": "recording_stopped" })) + } + _ => bail!("unsupported hosted worker command kind"), + } + } + + fn stop_generation_is_not_older(&self, guild_id: GuildId, generation: u64) -> bool { + let requested_is_newer = self + .requested_channels + .get(&guild_id) + .is_some_and(|requested| requested.generation > generation); + let active_is_newer = self + .active + .get(&guild_id) + .and_then(|active| active.hosted_generation) + .is_some_and(|active_generation| active_generation > generation); + !requested_is_newer && !active_is_newer } } +#[cfg(feature = "discord")] +fn hosted_guilds_without_heartbeat( + expected: &[(GuildId, String)], + heartbeated: Option<&std::collections::HashSet>, +) -> Vec { + expected + .iter() + .filter_map(|(guild_id, recording_id)| { + (!heartbeated.is_some_and(|ids| ids.contains(recording_id))).then_some(*guild_id) + }) + .collect() +} + #[cfg(feature = "discord")] fn capture_transition( active_channel: Option, @@ -1672,7 +3469,7 @@ fn capture_transition( #[async_trait] #[cfg(feature = "discord")] impl EventHandler for DiscordCaptureHandler { - async fn ready(&self, _: SerenityContext, ready: Ready) { + async fn ready(&self, ctx: SerenityContext, ready: Ready) { { let mut bot_user_id = self.bot_user_id.lock().expect("bot user id mutex poisoned"); *bot_user_id = Some(ready.user.id); @@ -1682,6 +3479,14 @@ impl EventHandler for DiscordCaptureHandler { ready.user.name, ready.user.id.get() ); + if let Some(hosted) = self.config.hosted.clone() + && !self.hosted_poller_started.swap(true, Ordering::SeqCst) + { + println!( + "Hosted worker mode enabled: occupancy auto-start is disabled; waiting for durable commands" + ); + tokio::spawn(self.clone().run_hosted_poller(ctx, hosted)); + } } async fn guild_create(&self, ctx: SerenityContext, guild: Guild, _: Option) { @@ -1945,21 +3750,99 @@ impl DiscordVoiceStats { } #[cfg(feature = "discord")] -fn create_wav_recorder(path: &Path) -> Result { - let recorder = SegmentedWavRecorder::new(path)?; - Ok(Arc::new(Mutex::new(Some(recorder)))) +fn create_wav_recorder(path: &Path) -> Result { + let recorder = SegmentedWavRecorder::new(path)?; + Ok(Arc::new(Mutex::new(Some(recorder)))) +} + +#[cfg(feature = "discord")] +fn finalize_wav(recorder: &SharedWavRecorder) -> Result> { + let recorder = { + let mut guard = recorder.lock().expect("wav recorder mutex poisoned"); + guard.take() + }; + if let Some(recorder) = recorder { + return recorder.finalize(); + } + Ok(Vec::new()) +} + +#[cfg(feature = "discord")] +fn checkpointed_wav_duration_seconds(base_path: &Path) -> Result { + let segments = checkpointed_wav_paths(base_path)?; + let expected_spec = discord_wav_spec(); + let mut total_samples = 0_u64; + for path in segments { + let metadata = std::fs::metadata(&path) + .with_context(|| format!("failed to inspect recovery WAV {}", path.display()))?; + if metadata.len() == 0 { + continue; + } + let reader = hound::WavReader::open(&path) + .with_context(|| format!("failed to read checkpointed WAV {}", path.display()))?; + if reader.spec() != expected_spec { + bail!("checkpointed WAV format did not match Discord capture format"); + } + total_samples = total_samples + .checked_add(u64::from(reader.duration())) + .context("checkpointed WAV duration overflowed")?; + } + Ok(total_samples.div_ceil(u64::from(DISCORD_SAMPLE_RATE))) +} + +#[cfg(feature = "discord")] +fn checkpointed_wav_paths(base_path: &Path) -> Result> { + let mut segments = vec![(1_u32, base_path.to_path_buf())]; + let mut missing_segment = false; + for index in 2..=MAX_HOSTED_RECOVERY_WAV_SEGMENTS { + let path = segment_wav_path(base_path, index); + if path.exists() { + if missing_segment { + bail!("hosted recovery WAV segments were not contiguous"); + } + segments.push((index, path)); + } else { + missing_segment = true; + } + } + if segment_wav_path(base_path, MAX_HOSTED_RECOVERY_WAV_SEGMENTS + 1).exists() { + bail!("hosted recovery WAV exceeded the bounded segment count"); + } + for (expected, (actual, _)) in (1_u32..).zip(&segments) { + if expected != *actual { + bail!("hosted recovery WAV segments were not contiguous"); + } + } + Ok(segments.into_iter().map(|(_, path)| path).collect()) +} + +#[cfg(feature = "discord")] +fn recovered_usage_seconds(base_path: &Path, reserved_seconds: u64) -> Result { + Ok(checkpointed_wav_duration_seconds(base_path)?.min(reserved_seconds)) } #[cfg(feature = "discord")] -fn finalize_wav(recorder: &SharedWavRecorder) -> Result> { - let recorder = { - let mut guard = recorder.lock().expect("wav recorder mutex poisoned"); - guard.take() - }; - if let Some(recorder) = recorder { - return recorder.finalize(); - } - Ok(Vec::new()) +fn authorized_usage_seconds( + started_at: DateTime, + authorization_ended_at: DateTime, + reserved_seconds: u64, +) -> u64 { + u64::try_from( + authorization_ended_at + .signed_duration_since(started_at) + .num_seconds() + .max(0), + ) + .unwrap_or(0) + .min(reserved_seconds) +} + +#[cfg(feature = "discord")] +fn hosted_authority_requires_fence(expires_at: DateTime, now: DateTime) -> bool { + expires_at + <= now + + chrono::Duration::from_std(RESERVATION_EXPIRY_MARGIN) + .expect("reservation expiry margin must fit chrono duration") } #[cfg(feature = "discord")] @@ -2941,6 +4824,51 @@ mod tests { use super::*; use serde_json::json; + #[cfg(feature = "discord")] + async fn isolated_test_pools() -> Result> { + let Ok(database_url) = std::env::var("CALL_SCRIBE_TEST_DATABASE_URL") else { + return Ok(None); + }; + let admin = PgPoolOptions::new() + .max_connections(1) + .connect(&database_url) + .await + .context("failed to connect to the disposable Postgres test database")?; + let schema = format!("call_scribe_test_{}", Uuid::new_v4().simple()); + sqlx::raw_sql::raw_sql(&format!("CREATE SCHEMA {schema}")) + .execute(&admin) + .await + .context("failed to create isolated Postgres test schema")?; + + async fn scoped_pool(database_url: &str, schema: String) -> Result { + PgPoolOptions::new() + .max_connections(2) + .after_connect(move |connection, _metadata| { + let statement = format!("SET search_path TO {schema}"); + Box::pin(async move { + sqlx::query::query(&statement).execute(connection).await?; + Ok(()) + }) + }) + .connect(database_url) + .await + .context("failed to connect to isolated Postgres test schema") + } + + let first = scoped_pool(&database_url, schema.clone()).await?; + let second = scoped_pool(&database_url, schema.clone()).await?; + Ok(Some((admin, first, second, schema))) + } + + #[cfg(feature = "discord")] + async fn drop_test_schema(admin: &PgPool, schema: &str) -> Result<()> { + sqlx::raw_sql::raw_sql(&format!("DROP SCHEMA {schema} CASCADE")) + .execute(admin) + .await + .context("failed to drop isolated Postgres test schema")?; + Ok(()) + } + #[cfg(feature = "discord")] fn discord_capture_handler_for_test( allowed_channel_id: Option, @@ -2954,6 +4882,60 @@ mod tests { allowed_channel_id, runtime_store: None, session_tx, + hosted: None, + self_hosted_organization_id: DEFAULT_ORGANIZATION_ID.to_string(), + self_hosted_capture_mode: CaptureMode::RecordOnly, + }) + } + + #[cfg(feature = "discord")] + fn hosted_capture_handler_for_test() -> DiscordCaptureHandler { + let (session_tx, _session_rx) = tokio::sync::mpsc::channel(1); + let configurations = HostedConfigurationStore::new(Duration::from_secs(60)); + configurations.replace(hosted_control::GuildConfigurationResponse { + revision: "r1".to_string(), + guilds: vec![hosted_control::GuildConfiguration { + guild_id: "1".to_string(), + organization_id: "org_1".to_string(), + entitlement_active: true, + approved_channel_ids: vec!["2".to_string()], + notice_channel_id: Some("3".to_string()), + consent_mode: Some("explicit_command".to_string()), + consent_policy_version: Some("v1".to_string()), + consent_notice_template: Some( + "This call will be recorded after a durable start command.".to_string(), + ), + retention_days: Some(30), + recording_enabled: true, + monthly_recording_seconds_cap: Some(3_600), + remaining_recording_seconds: Some(300), + storage_provider: Some("customer_s3".to_string()), + storage_destination_label: Some("worker-pvc".to_string()), + ready: true, + blocked_reasons: Vec::new(), + desired_recording_generation: 1, + }], + }); + DiscordCaptureHandler::new(DiscordCaptureConfig { + capture_dir: PathBuf::from("data/test-captures"), + trigger_user_id: None, + guild_id: None, + allowed_channel_id: None, + runtime_store: None, + session_tx, + hosted: Some(HostedCaptureConfig { + client: HostedControlPlaneClient::new( + "http://127.0.0.1:8080", + "test-token-with-at-least-thirty-two-bytes".to_string(), + "test-worker".to_string(), + "test-outbox-key-with-at-least-thirty-two-bytes".to_string(), + ) + .expect("loopback hosted client should be valid"), + configurations, + poll_interval: Duration::from_secs(15), + }), + self_hosted_organization_id: DEFAULT_ORGANIZATION_ID.to_string(), + self_hosted_capture_mode: CaptureMode::RecordOnly, }) } @@ -3057,6 +5039,97 @@ mod tests { assert_eq!(handler.desired_capture_channel(guild_id), None); } + #[cfg(feature = "discord")] + #[test] + fn hosted_mode_stays_closed_after_command_until_storage_adapter_exists() { + let guild_id = GuildId::new(1); + let channel_id = ChannelId::new(2); + let participant_user_id = UserId::new(4); + let handler = hosted_capture_handler_for_test(); + handler + .voice_states + .insert((guild_id, participant_user_id), Some(channel_id)); + + assert_eq!(handler.desired_capture_channel(guild_id), None); + handler.requested_channels.insert( + guild_id, + RequestedCapture { + channel_id, + generation: 1, + command_id: "cmd-1".to_string(), + }, + ); + assert_eq!(handler.desired_capture_channel(guild_id), None); + } + + #[cfg(feature = "discord")] + #[test] + fn hosted_mode_denies_unapproved_explicit_channel() { + let guild_id = GuildId::new(1); + let approved_channel_id = ChannelId::new(2); + let unapproved_channel_id = ChannelId::new(9); + let participant_user_id = UserId::new(4); + let handler = hosted_capture_handler_for_test(); + handler + .voice_states + .insert((guild_id, participant_user_id), Some(unapproved_channel_id)); + handler.requested_channels.insert( + guild_id, + RequestedCapture { + channel_id: unapproved_channel_id, + generation: 1, + command_id: "cmd-1".to_string(), + }, + ); + + assert_eq!(handler.desired_capture_channel(guild_id), None); + handler + .voice_states + .insert((guild_id, participant_user_id), Some(approved_channel_id)); + assert_eq!(handler.desired_capture_channel(guild_id), None); + } + + #[cfg(feature = "discord")] + #[test] + fn hosted_mode_rejects_a_requested_generation_older_than_policy() { + let guild_id = GuildId::new(1); + let channel_id = ChannelId::new(2); + let participant_user_id = UserId::new(4); + let handler = hosted_capture_handler_for_test(); + handler + .voice_states + .insert((guild_id, participant_user_id), Some(channel_id)); + handler.requested_channels.insert( + guild_id, + RequestedCapture { + channel_id, + generation: 0, + command_id: "cmd-0".to_string(), + }, + ); + + assert_eq!(handler.desired_capture_channel(guild_id), None); + } + + #[cfg(feature = "discord")] + #[test] + fn stale_stop_generation_cannot_clear_newer_requested_start() { + let guild_id = GuildId::new(1); + let channel_id = ChannelId::new(2); + let handler = hosted_capture_handler_for_test(); + handler.requested_channels.insert( + guild_id, + RequestedCapture { + channel_id, + generation: 2, + command_id: "cmd-2".to_string(), + }, + ); + + assert!(!handler.stop_generation_is_not_older(guild_id, 1)); + assert!(handler.stop_generation_is_not_older(guild_id, 2)); + } + #[cfg(feature = "discord")] #[tokio::test] async fn join_leave_rejoin_leave_during_start_requests_stop() { @@ -3068,7 +5141,8 @@ mod tests { .voice_states .insert((guild_id, participant_user_id), Some(channel_id)); - let starting_reconcile = handler.reconcile_gate.lock().await; + let starting_gate = handler.reconcile_gate_for(guild_id); + let starting_reconcile = starting_gate.lock().await; handler .voice_states .insert((guild_id, participant_user_id), None); @@ -3080,7 +5154,8 @@ mod tests { waiting_handler .voice_states .insert((guild_id, participant_user_id), None); - let _waiting_reconcile = waiting_handler.reconcile_gate.lock().await; + let waiting_gate = waiting_handler.reconcile_gate_for(guild_id); + let _waiting_reconcile = waiting_gate.lock().await; capture_transition( Some(channel_id), waiting_handler.desired_capture_channel(guild_id), @@ -3104,13 +5179,15 @@ mod tests { let participant_user_id = UserId::new(3); let handler = discord_capture_handler_for_test(Some(channel_id), None); - let stopping_reconcile = handler.reconcile_gate.lock().await; + let stopping_gate = handler.reconcile_gate_for(guild_id); + let stopping_reconcile = stopping_gate.lock().await; let waiting_handler = handler.clone(); let rejoin = tokio::spawn(async move { waiting_handler .voice_states .insert((guild_id, participant_user_id), Some(channel_id)); - let _waiting_reconcile = waiting_handler.reconcile_gate.lock().await; + let waiting_gate = waiting_handler.reconcile_gate_for(guild_id); + let _waiting_reconcile = waiting_gate.lock().await; capture_transition(None, waiting_handler.desired_capture_channel(guild_id)) }); @@ -3122,4 +5199,424 @@ mod tests { CaptureTransition::Start(channel_id) ); } + + #[cfg(feature = "discord")] + #[tokio::test] + async fn guild_transition_locks_do_not_block_other_guilds() { + let handler = discord_capture_handler_for_test(None, None); + let first_gate = handler.reconcile_gate_for(GuildId::new(1)); + let _first_guard = first_gate.lock().await; + let second_gate = handler.reconcile_gate_for(GuildId::new(2)); + + assert!(second_gate.try_lock().is_ok()); + } + + #[cfg(feature = "discord")] + #[test] + fn crash_recovery_uses_only_checkpointed_audio_duration() { + let test_dir = + std::env::temp_dir().join(format!("call-scribe-crash-duration-{}", Uuid::new_v4())); + std::fs::create_dir_all(&test_dir).expect("test capture directory should be created"); + let wav_path = test_dir.join("capture.wav"); + let mut writer = hound::WavWriter::create(&wav_path, discord_wav_spec()) + .expect("test WAV should be created"); + let samples_per_second = + usize::try_from(DISCORD_SAMPLE_RATE).unwrap() * usize::from(DISCORD_CHANNELS); + for _ in 0..(2 * samples_per_second) { + writer.write_sample(0_i16).expect("sample should write"); + } + writer.flush().expect("two seconds should be checkpointed"); + for _ in 0..samples_per_second { + writer + .write_sample(0_i16) + .expect("uncheckpointed sample should write"); + } + + // Simulate abrupt process loss: do not drop/finalize the writer. The + // recovery boundary may bill only the valid duration in the last + // checkpointed WAV header, never started_at-to-now wall time. + std::mem::forget(writer); + assert_eq!( + checkpointed_wav_duration_seconds(&wav_path) + .expect("checkpointed duration should be recoverable"), + 2 + ); + assert_eq!( + recovered_usage_seconds(&wav_path, 1) + .expect("recovered usage must remain reservation-bounded"), + 1 + ); + + let skipped_segment_path = segment_wav_path(&wav_path, 3); + hound::WavWriter::create(&skipped_segment_path, discord_wav_spec()) + .expect("skipped test WAV should be created") + .finalize() + .expect("skipped test WAV should finalize"); + assert!( + checkpointed_wav_duration_seconds(&wav_path).is_err(), + "a missing segment must stay pending instead of undercounting" + ); + + std::fs::remove_dir_all(&test_dir).expect("test capture directory should be removed"); + } + + #[cfg(feature = "discord")] + #[tokio::test] + async fn hosted_outbox_upgrade_migration_clears_terminal_lease_material() -> Result<()> { + let Some((admin, first, _second, schema)) = isolated_test_pools().await? else { + eprintln!("CALL_SCRIBE_TEST_DATABASE_URL is unset; skipping Postgres migration proof"); + return Ok(()); + }; + + sqlx::raw_sql::raw_sql(include_str!( + "../migrations/20260812010000_hosted_worker_command_executions.sql" + )) + .execute(&first) + .await?; + sqlx::query::query( + r#" +INSERT INTO call_scribe_hosted_usage_outbox + (reservation_id, encrypted_lease_token, encryption_nonce, recording_id, + actual_seconds, occurred_at, expires_at, status) +VALUES ('reservation-upgrade', '\x01', '\x02', 'recording-upgrade', + 1, now(), now() + interval '1 hour', 'delivered') +"#, + ) + .execute(&first) + .await?; + + migrate_runtime_schema(&first).await?; + migrate_runtime_schema(&first).await?; + + let terminal_material: (Option>, Option>) = sqlx::query_as::query_as( + r#" +SELECT encrypted_lease_token, encryption_nonce +FROM call_scribe_hosted_usage_outbox +WHERE reservation_id = 'reservation-upgrade' +"#, + ) + .fetch_one(&first) + .await?; + assert_eq!(terminal_material, (None, None)); + + let invalid_pending = sqlx::query::query( + r#" +INSERT INTO call_scribe_hosted_usage_outbox + (reservation_id, encrypted_lease_token, encryption_nonce, recording_id, + actual_seconds, occurred_at, expires_at, status) +VALUES ('reservation-invalid', NULL, NULL, 'recording-invalid', + 1, now(), now() + interval '1 hour', 'pending') +"#, + ) + .execute(&first) + .await; + assert!( + invalid_pending.is_err(), + "upgraded outbox must reject pending rows without encrypted lease material" + ); + + first.close().await; + drop_test_schema(&admin, &schema).await?; + admin.close().await; + Ok(()) + } + + #[cfg(feature = "discord")] + #[tokio::test] + async fn two_instances_cannot_recover_a_live_heartbeating_capture() -> Result<()> { + let Some((admin, first, second, schema)) = isolated_test_pools().await? else { + eprintln!("CALL_SCRIBE_TEST_DATABASE_URL is unset; skipping Postgres replica proof"); + return Ok(()); + }; + migrate_runtime_schema(&first).await?; + let first_store = SqlxRuntimeStore { + pool: first.clone(), + organization_id: "org-test".to_string(), + capture_mode: CaptureMode::RecordOnly, + }; + let second_store = SqlxRuntimeStore { + pool: second.clone(), + organization_id: "org-test".to_string(), + capture_mode: CaptureMode::RecordOnly, + }; + + sqlx::query::query( + r#" +INSERT INTO call_scribe_hosted_capture_recovery + (reservation_id, encrypted_lease_token, encryption_nonce, recording_id, + base_wav_path, reserved_seconds, started_at, expires_at, owner_instance_id) +VALUES ('reservation-live', '\x01', '\x02', 'recording-live', + '/captures/live.wav', 300, now() - interval '1 minute', + now() + interval '1 hour', 'instance-a') +"#, + ) + .execute(&first) + .await?; + + let second_claim = second_store + .claim_abandoned_hosted_capture_recoveries(Duration::from_secs(60), "claim-b") + .await?; + assert!( + second_claim.is_empty(), + "instance B must not recover instance A's fresh live capture" + ); + + let renewed_expires_at = Utc::now() + chrono::Duration::minutes(90); + first_store + .renew_hosted_capture_recovery( + "instance-a", + "reservation-live", + "recording-live", + renewed_expires_at, + ) + .await?; + let persisted_expires_at: DateTime = sqlx::query_scalar::query_scalar( + r#"SELECT expires_at FROM call_scribe_hosted_capture_recovery + WHERE reservation_id = 'reservation-live'"#, + ) + .fetch_one(&second) + .await?; + assert!( + persisted_expires_at + .signed_duration_since(renewed_expires_at) + .num_milliseconds() + .abs() + <= 1 + ); + assert!( + second_store + .claim_abandoned_hosted_capture_recoveries(Duration::from_secs(60), "claim-b") + .await? + .is_empty(), + "a renewed DB heartbeat must remain authoritative across pools" + ); + + sqlx::query::query( + r#" +UPDATE call_scribe_hosted_capture_recovery +SET heartbeat_at = now() - interval '2 minutes' +WHERE reservation_id = 'reservation-live' +"#, + ) + .execute(&first) + .await?; + let abandoned = second_store + .claim_abandoned_hosted_capture_recoveries(Duration::from_secs(60), "claim-b") + .await?; + assert_eq!( + abandoned.len(), + 1, + "stale ownership must become recoverable" + ); + assert!( + first_store + .renew_hosted_capture_recovery( + "instance-a", + "reservation-live", + "recording-live", + Utc::now() + chrono::Duration::minutes(90), + ) + .await + .is_err(), + "the previous owner must not renew a recovery after it was fenced" + ); + assert!( + first_store + .remove_live_hosted_capture_recovery( + "reservation-live", + "recording-live", + "instance-a", + ) + .await + .is_err(), + "the previous owner must not delete a claimed recovery" + ); + assert!( + second_store + .remove_claimed_hosted_capture_recovery( + "reservation-live", + "recording-live", + "wrong-claim", + ) + .await + .is_err(), + "a stale recovery claimant must not cross the claim-token fence" + ); + second_store + .remove_claimed_hosted_capture_recovery("reservation-live", "recording-live", "claim-b") + .await?; + + first.close().await; + second.close().await; + drop_test_schema(&admin, &schema).await?; + admin.close().await; + Ok(()) + } + + #[cfg(feature = "discord")] + #[test] + fn missing_or_failed_heartbeat_self_fences_only_unowned_guilds() { + let expected = vec![ + (GuildId::new(1), "recording-a".to_string()), + (GuildId::new(2), "recording-b".to_string()), + ]; + let renewed = std::collections::HashSet::from(["recording-a".to_string()]); + + assert_eq!( + hosted_guilds_without_heartbeat(&expected, Some(&renewed)), + vec![GuildId::new(2)] + ); + assert_eq!( + hosted_guilds_without_heartbeat(&expected, None), + vec![GuildId::new(1), GuildId::new(2)], + "a heartbeat query failure must self-fence every hosted capture" + ); + } + + #[cfg(feature = "discord")] + #[test] + fn final_usage_is_bounded_by_authoritative_capture_end() { + let started_at = Utc::now(); + assert_eq!( + authorized_usage_seconds(started_at, started_at + chrono::Duration::seconds(25), 300,), + 25 + ); + assert_eq!( + authorized_usage_seconds(started_at, started_at + chrono::Duration::seconds(500), 300,), + 300 + ); + assert_eq!( + authorized_usage_seconds(started_at, started_at - chrono::Duration::seconds(1), 300,), + 0 + ); + } + + #[cfg(feature = "discord")] + #[test] + fn local_authority_watchdog_fences_at_safety_margin() { + let now = Utc::now(); + assert!(!hosted_authority_requires_fence( + now + chrono::Duration::seconds(16), + now, + )); + assert!(hosted_authority_requires_fence( + now + chrono::Duration::seconds(15), + now, + )); + assert!(hosted_authority_requires_fence( + now - chrono::Duration::seconds(1), + now, + )); + } + + #[cfg(feature = "discord")] + #[tokio::test] + async fn heartbeat_shortening_is_durable_and_final_usage_uses_settlement_window() -> Result<()> + { + let Some((admin, first, _second, schema)) = isolated_test_pools().await? else { + eprintln!("CALL_SCRIBE_TEST_DATABASE_URL is unset; skipping Postgres heartbeat proof"); + return Ok(()); + }; + migrate_runtime_schema(&first).await?; + let store = SqlxRuntimeStore { + pool: first.clone(), + organization_id: "org-test".to_string(), + capture_mode: CaptureMode::RecordOnly, + }; + let client = HostedControlPlaneClient::new( + "http://127.0.0.1:8080", + "test-secret-with-at-least-thirty-two-bytes".to_string(), + "worker-1".to_string(), + "test-outbox-secret-with-at-least-thirty-two-bytes".to_string(), + )?; + let started_at = Utc::now() - chrono::Duration::seconds(20); + let initial_expiry = Utc::now() + chrono::Duration::seconds(80); + let reservation = UsageReservation { + reservation_id: "reservation-shortened".to_string(), + lease_token: "opaque-lease-token".to_string(), + reserved_seconds: 300, + expires_at: initial_expiry.to_rfc3339(), + }; + let base_wav_path = Path::new("/captures/shortened.wav"); + store + .persist_hosted_capture_recovery( + &client, + &reservation, + "recording-shortened", + base_wav_path, + started_at.with_timezone(&Local), + "instance-a", + ) + .await?; + + let shortened_expiry = Utc::now() + chrono::Duration::seconds(40); + store + .renew_hosted_capture_recovery( + "instance-a", + "reservation-shortened", + "recording-shortened", + shortened_expiry, + ) + .await?; + let (claim_token, authorization_end) = store + .claim_live_hosted_capture_finalization( + "reservation-shortened", + "recording-shortened", + "instance-a", + Utc::now(), + ) + .await?; + assert!(authorization_end <= shortened_expiry); + + let actual_seconds = authorized_usage_seconds(started_at, authorization_end, 300); + assert!((19..=21).contains(&actual_seconds)); + let mut settlement_reservation = reservation; + settlement_reservation.expires_at = authorization_end.to_rfc3339(); + store + .enqueue_hosted_usage( + &client, + &settlement_reservation, + "recording-shortened", + actual_seconds, + started_at + chrono::Duration::seconds(i64::try_from(actual_seconds)?), + ) + .await?; + store + .remove_claimed_hosted_capture_recovery( + "reservation-shortened", + "recording-shortened", + &claim_token, + ) + .await?; + + let (queued_seconds, queued_expires_at): (i64, DateTime) = sqlx::query_as::query_as( + r#"SELECT actual_seconds, expires_at + FROM call_scribe_hosted_usage_outbox + WHERE reservation_id = 'reservation-shortened'"#, + ) + .fetch_one(&first) + .await?; + assert_eq!(queued_seconds, i64::try_from(actual_seconds)?); + assert!( + queued_expires_at + >= authorization_end + chrono::Duration::minutes(30) + - chrono::Duration::milliseconds(1) + ); + + first.close().await; + drop_test_schema(&admin, &schema).await?; + admin.close().await; + Ok(()) + } + + #[cfg(feature = "discord")] + #[test] + fn hosted_start_revalidates_ownership_before_stale_takeover() { + let maximum_pre_activation_window = + DISCORD_VOICE_TRANSITION_TIMEOUT + HOSTED_START_STEP_TIMEOUT.saturating_mul(3); + assert!( + maximum_pre_activation_window < MIN_HOSTED_RECOVERY_STALE_AFTER, + "join, session persistence, handler acquisition, and final ownership renewal must finish before another replica may claim the row" + ); + } }