diff --git a/.changes/018-video-extract-audio.md b/.changes/018-video-extract-audio.md new file mode 100644 index 000000000..fbb9d26e0 --- /dev/null +++ b/.changes/018-video-extract-audio.md @@ -0,0 +1,32 @@ +--- +group: platform +component: Integrations +impact: compatible +category: Added +--- + +## Summary + +You can now pull the audio out of an uploaded video with the new +`Integration.Video.ExtractAudio` integration. Give it the video's file +reference and it returns the audio as a mono 16 kHz WAV (`audio/wav`), ready +to pass to `Integration.Audio.Transcribe` with `mimeType = "audio/wav"`. By +default it allows 180 seconds and stops at 25 MB; both limits are in the +config. It needs `ffmpeg` on the machine that runs your app (for example +`brew install ffmpeg`). If `ffmpeg` is missing, the action fails with a +message that tells you to install it. If the video itself cannot be +converted, your `onError` command receives the reason. + +Raise the dispatcher's `eventProcessingTimeoutMs` above the extraction +timeout, for example to `Just 200000`. Its 30-second default would otherwise +stop a slow extraction before your `onSuccess` or `onError` command is sent. +To keep the audio in a command, store it as Base64 text +(`result.audio |> Bytes.toBase64 |> Text.fromBytes`). The new +`Bytes.fromBase64` turns it back into the exact audio bytes +(`audioBase64 |> Text.toBytes |> Bytes.fromBase64`) and returns an error +instead of different bytes if the text is not valid Base64. + +Existing applications do not need any changes. To verify locally, run +`./dev test 'Integration.Video' nhintegrations-test` and confirm the +extraction examples pass. They use a stand-in for `ffmpeg`, so you do not +need it installed to run them. diff --git a/.hlint.yaml b/.hlint.yaml index fec28a471..e9b4842cc 100644 --- a/.hlint.yaml +++ b/.hlint.yaml @@ -62,8 +62,10 @@ within: [Command, Config, Console, Environment, Log, Main, Service.Integration.Selection], message: "Use Core `Environment`. New exceptions need a justification + belongs-in note here."} # Directory/Path + # Integration.Video.ExtractAudio.Internal: needs the OS temp directory for + # the video/WAV temp files; belongs-in: Core `Directory` (getTemporary). - {name: [System.Directory], - within: [Directory, Environment, File, Integration.Pdf.ExtractText.Internal, Subprocess], + within: [Directory, Environment, File, Integration.Pdf.ExtractText.Internal, Integration.Video.ExtractAudio.Internal, Subprocess], message: "Use Core `Directory`/`Path`. New exceptions need a justification + belongs-in note here."} # Json - {name: [Data.Aeson, Data.Aeson.Key, Data.Aeson.KeyMap, Data.Aeson.Types, Data.Aeson.Encoding], diff --git a/codemap/api-hot.md b/codemap/api-hot.md index c2b8e1dd7..0f4a16cbc 100644 --- a/codemap/api-hot.md +++ b/codemap/api-hot.md @@ -10,12 +10,12 @@ codemap/signatures/ · type-directed search: ./dev api "" ## Task -- `yield :: value -> Task w value` -- `ignoreError :: Task err Unit -> Task w Unit` -- `mapError :: (err1 -> err2) -> Task err1 value -> Task err2 value` -- `throw :: err -> Task err w` -- `fromIO :: IO value -> Task w value` -- `asResult :: Task err value -> Task err2 (Result err value)` +- `yield :: value -> Task w value` +- `ignoreError :: Task err Unit -> Task w Unit` +- `mapError :: (err1 -> err2) -> Task err1 value -> Task err2 value` +- `throw :: err -> Task err w` +- `fromIO :: IO value -> Task w value` +- `asResult :: Task err value -> Task err2 (Result err value)` - `when :: Bool -> Task err Unit -> Task err Unit` - `forEach :: (element -> Task err Unit) -> Array element -> Task err Unit` - `fromFailableIO :: Exception exception => IO result -> Task exception result` @@ -25,7 +25,7 @@ codemap/signatures/ · type-directed search: ./dev api "" - `map :: (a -> b) -> Array a -> Array b` - `>>> map sqrt (fromLinkedList [1,4,9] :: Array Float)` → `Array [1.0,2.0,3.0]` -- `fromLinkedList :: LinkedList a -> Array a` +- `fromLinkedList :: LinkedList a -> Array a` - `empty :: Array a` - `>>> empty :: Array Int` → `Array []` - `toLinkedList :: Array a -> LinkedList a` @@ -45,12 +45,12 @@ codemap/signatures/ · type-directed search: ./dev api "" ## Text -- `fromLinkedList :: LinkedList Char -> Text` +- `fromLinkedList :: LinkedList Char -> Text` - `toLinkedList :: Text -> LinkedList Char` - `toBytes :: Text -> Bytes` - `replace :: Text -> Text -> Text -> Text` +- `trim :: Text -> Text` - `fromBytes :: Bytes -> Text` -- `trim :: Text -> Text` - `joinWith :: Text -> Array Text -> Text` - `isEmpty :: Text -> Bool` - `append :: Text -> Text -> Text` @@ -130,16 +130,16 @@ codemap/signatures/ · type-directed search: ./dev api "" ## Maybe -- `withDefault :: a -> Maybe a -> a` +- `withDefault :: a -> Maybe a -> a` - `map :: (a -> b) -> Maybe a -> Maybe b` - `getOrDie :: HasCallStack => Maybe a -> a` - `andThen :: (a -> Maybe b) -> Maybe a -> Maybe b` ## Integration -- `action :: (ActionContext -> Task IntegrationError (Maybe CommandPayload)) -> Action` +- `action :: (ActionContext -> Task IntegrationError (Maybe CommandPayload)) -> Action` - `outbound :: ToAction config => config -> Action` -- `emitCommand :: forall command (name :: Symbol). (ToJSON command, name ~ NameOf command, KnownSymbol name) => command -> Task IntegrationError (Maybe CommandPayload)` +- `emitCommand :: forall command (name :: Symbol). (ToJSON command, name ~ NameOf command, KnownSymbol name) => command -> Task IntegrationError (Maybe CommandPayload)` - `toAction :: ToAction config => config -> Action` - `fromMap :: Map Text ValidatedOAuth2ProviderConfig -> ImmutableProviderRegistry` - `runAction :: ActionContext -> Action -> Task IntegrationError (Maybe CommandPayload)` @@ -167,6 +167,17 @@ codemap/signatures/ · type-directed search: ./dev api "" - `load :: HasParser config => Task Text config` - `secret :: FieldDef -> FieldDef` +## Uuid + +- `generate :: Task w Uuid` +- `toText :: Uuid -> Text` +- `nil :: Uuid` +- `toLegacy :: Uuid -> UUID` +- `fromText :: Text -> Maybe Uuid` +- `generateV5 :: Uuid -> Text -> Uuid` + - `>>> Uuid.fromText "6ba7b810-9dad-11d1-80b4-00c04fd430c8" |> Maybe.getOrDie |> (\ns -> Uuid.generateV5 ns "python.org") |> Uuid.toText` → `"886313e1-3b8a-5372-9b90-0c9aee199e5d"` +- `fromLegacy :: UUID -> Uuid` + ## LinkedList - `map :: (a -> b) -> LinkedList a -> LinkedList b` @@ -180,17 +191,6 @@ codemap/signatures/ · type-directed search: ./dev api "" - `all :: (a -> Bool) -> LinkedList a -> Bool` - `concat :: LinkedList (LinkedList a) -> LinkedList a` -## Uuid - -- `generate :: Task w Uuid` -- `toText :: Uuid -> Text` -- `nil :: Uuid` -- `toLegacy :: Uuid -> UUID` -- `fromText :: Text -> Maybe Uuid` -- `generateV5 :: Uuid -> Text -> Uuid` - - `>>> Uuid.fromText "6ba7b810-9dad-11d1-80b4-00c04fd430c8" |> Maybe.getOrDie |> (\ns -> Uuid.generateV5 ns "python.org") |> Uuid.toText` → `"886313e1-3b8a-5372-9b90-0c9aee199e5d"` -- `fromLegacy :: UUID -> Uuid` - ## AsyncTask - `sleep :: Int -> Task w Unit` @@ -204,11 +204,11 @@ codemap/signatures/ · type-directed search: ./dev api "" ## Path - `toLinkedList :: Path -> LinkedList Char` -- `fromText :: Text -> Maybe Path` -- `toText :: Path -> Text` -- `append :: Path -> Path -> Path` +- `fromText :: Text -> Maybe Path` +- `toText :: Path -> Text` +- `append :: Path -> Path -> Path` +- `fromLinkedList :: LinkedList Char -> Maybe Path` - `joinPaths :: Array Path -> Path` -- `fromLinkedList :: LinkedList Char -> Maybe Path` - `endsWith :: Text -> Path -> Bool` ## ConcurrentMap @@ -283,7 +283,7 @@ codemap/signatures/ · type-directed search: ./dev api "" - `fmap :: Functor f => (a -> b) -> f a -> f b` - `map :: Mappable mappable => (typeA -> typeB) -> mappable typeA -> mappable typeB` -*cut: 103 more modules (AtomicVar (26), Integration.Http (26), Decider (21), Result (21), Lock (20), …) — full surface: codemap/signatures/* +*cut: 103 more modules (AtomicVar (26), Integration.Http (26), Result (24), Decider (21), Lock (20), …) — full surface: codemap/signatures/* --- @@ -294,35 +294,35 @@ when WRITING TESTS; they are not feature-code frequency signal. ## Task -- `mapError :: (err1 -> err2) -> Task err1 value -> Task err2 value` -- `yield :: value -> Task w value` -- `asResult :: Task err value -> Task err2 (Result err value)` -- `mapArray :: (element -> Task err output) -> Array element -> Task err (Array output)` -- `throw :: err -> Task err w` +- `mapError :: (err1 -> err2) -> Task err1 value -> Task err2 value` +- `yield :: value -> Task w value` +- `asResult :: Task err value -> Task err2 (Result err value)` +- `mapArray :: (element -> Task err output) -> Array element -> Task err (Array output)` +- `throw :: err -> Task err w` - `andThen :: (input -> Task err output) -> Task err input -> Task err output` - `fromIO :: IO value -> Task w value` -- `map :: (input -> output) -> Task err input -> Task err output` +- `map :: (input -> output) -> Task err input -> Task err output` ## Array -- `length :: Array a -> Int` -- `empty :: Array a` -- `fromLinkedList :: LinkedList a -> Array a` -- `map :: (a -> b) -> Array a -> Array b` -- `get :: Int -> Array a -> Maybe a` -- `wrap :: a -> Array a` +- `length :: Array a -> Int` +- `empty :: Array a` +- `fromLinkedList :: LinkedList a -> Array a` +- `map :: (a -> b) -> Array a -> Array b` +- `get :: Int -> Array a -> Maybe a` +- `wrap :: a -> Array a` - `initialize :: Int -> (Int -> a) -> Array a` - `contains :: Eq value => value -> Array value -> Bool` ## Text -- `contains :: Text -> Text -> Bool` -- `toBytes :: Text -> Bytes` +- `contains :: Text -> Text -> Bool` +- `toBytes :: Text -> Bytes` - `fromLinkedList :: LinkedList Char -> Text` - `length :: Text -> Int` - `repeat :: Int -> Text -> Text` - `toLinkedList :: Text -> LinkedList Char` -- `startsWith :: Text -> Text -> Bool` +- `startsWith :: Text -> Text -> Bool` - `all :: (Char -> Bool) -> Text -> Bool` ## Parser @@ -340,11 +340,11 @@ when WRITING TESTS; they are not feature-code frequency signal. - `toJSON :: ToJSON a => a -> Value` - `decodeText :: FromJSON value => Text -> Result Text value` -- `encodeText :: ToJSON value => value -> Text` +- `encodeText :: ToJSON value => value -> Text` - `object :: [(Text, Value)] -> Value` - `null :: Value` -- `encode :: ToJSON value => value -> Value` -- `decode :: FromJSON value => Value -> Result Text value` +- `encode :: ToJSON value => value -> Value` +- `decode :: FromJSON value => Value -> Result Text value` - `yield :: value -> Parser value` ## Layout @@ -360,9 +360,9 @@ when WRITING TESTS; they are not feature-code frequency signal. ## ConcurrentVar -- `containing :: value -> Task w (ConcurrentVar value)` -- `modify :: (value -> value) -> ConcurrentVar value -> Task w Unit` -- `peek :: ConcurrentVar value -> Task w value` +- `containing :: value -> Task w (ConcurrentVar value)` +- `modify :: (value -> value) -> ConcurrentVar value -> Task w Unit` +- `peek :: ConcurrentVar value -> Task w value` - `get :: ConcurrentVar value -> Task w value` - `set :: value -> ConcurrentVar value -> Task w ()` - `new :: forall value w. Task w (ConcurrentVar value)` @@ -372,9 +372,9 @@ when WRITING TESTS; they are not feature-code frequency signal. ## Uuid - `generate :: Task w Uuid` -- `nil :: Uuid` +- `nil :: Uuid` - `toText :: Uuid -> Text` - `generateV5 :: Uuid -> Text -> Uuid` - `fromText :: Text -> Maybe Uuid` -*cut: 138 more modules (Map (257), Service.Application (250), Set (233), AsyncTask (223), Stream (212), …) — full surface: codemap/signatures/* +*cut: 140 more modules (Map (260), Service.Application (250), Set (233), AsyncTask (225), Stream (212), …) — full surface: codemap/signatures/* diff --git a/codemap/phrasebook.md b/codemap/phrasebook.md index 5710950a9..69ab67c04 100644 --- a/codemap/phrasebook.md +++ b/codemap/phrasebook.md @@ -89,6 +89,16 @@ Array [0] [] ``` +```haskell +>>> "aGVsbG8=" |> Text.toBytes |> Bytes.fromBase64 |> Result.map Text.fromBytes +Ok "hello" +``` + +```haskell +>>> "aGVsbG8" |> Text.toBytes |> Bytes.fromBase64 |> Result.isErr +True +``` + ### Float ```haskell @@ -194,6 +204,29 @@ Set (fromList [1,2,3]) "Cart" ``` +## outbound-integrations + +### Integration.Video.ExtractAudio + +```haskell +>>> defaultConfig.timeoutSeconds +180 +``` + +```haskell +>>> defaultConfig.sampleRateHz +16000 +``` + +### Integration.Video.ExtractAudio.Internal + +```haskell +>>> import Array qualified +>>> import Integration.Video.ExtractAudio (defaultConfig) +>>> Array.toLinkedList (buildFfmpegArgs defaultConfig "in.mp4" "out.wav") +["-nostdin","-v","error","-i","in.mp4","-vn","-ac","1","-ar","16000","-fs","25000000","out.wav"] +``` + ## postgres-infra ### Service.Infra.Postgres.SslMode @@ -237,4 +270,4 @@ True --- -*32 example sessions · modules with ZERO doctests: 315 (the documentation backlog — see codemap/.doc-ratchet)* +*37 example sessions · modules with ZERO doctests: 315 (the documentation backlog — see codemap/.doc-ratchet)* diff --git a/codemap/signatures/nhcore-core.txt b/codemap/signatures/nhcore-core.txt index acf6d8f47..f085a6499 100644 --- a/codemap/signatures/nhcore-core.txt +++ b/codemap/signatures/nhcore-core.txt @@ -275,6 +275,7 @@ drop :: Int -> Bytes -> Bytes dropEnd :: Int -> Bytes -> Bytes empty :: Bytes findSubstring :: Bytes -> Bytes -> Maybe Int +fromBase64 :: Bytes -> Result Text Bytes fromLazyLegacy :: LazyByteString -> Bytes fromLegacy :: ByteString -> Bytes getRandom :: Int -> Task w Bytes diff --git a/codemap/signatures/nhintegrations.txt b/codemap/signatures/nhintegrations.txt index 86518882d..45da8ecb6 100644 --- a/codemap/signatures/nhintegrations.txt +++ b/codemap/signatures/nhintegrations.txt @@ -1881,3 +1881,42 @@ buildPdftotextArgs :: Config -> Path -> Array Text extractPageCount :: Text -> Int instance (Data.Aeson.Types.ToJSON.ToJSON command, GHC.TypeLits.KnownSymbol (Service.Command.Core.NameOf command)) => Integration.ToAction (Integration.Pdf.ExtractText.Request command) parsePdfInfo :: Text -> PdfMetadata + +module Integration.Video.ExtractAudio +Config :: Int -> Int -> Int -> Int -> Maybe Text -> Config +ExtractionResult :: Bytes -> Text -> ExtractionResult +Request :: FileRef -> Config -> (ExtractionResult -> command) -> (Text -> command) -> Request command +[audio] :: ExtractionResult -> Bytes +[channels] :: Config -> Int +[config] :: Request command -> Config +[ffmpegPath] :: Config -> Maybe Text +[fileRef] :: Request command -> FileRef +[maxOutputBytes] :: Config -> Int +[mimeType] :: ExtractionResult -> Text +[onError] :: Request command -> Text -> command +[onSuccess] :: Request command -> ExtractionResult -> command +[sampleRateHz] :: Config -> Int +[timeoutSeconds] :: Config -> Int +data Config +data ExtractionResult +data Request command +defaultConfig :: Config +instance GHC.Classes.Eq Integration.Video.ExtractAudio.Config +instance GHC.Classes.Eq Integration.Video.ExtractAudio.ExtractionResult +instance GHC.Generics.Generic (Integration.Video.ExtractAudio.Request command) +instance GHC.Generics.Generic Integration.Video.ExtractAudio.Config +instance GHC.Generics.Generic Integration.Video.ExtractAudio.ExtractionResult +instance GHC.Show.Show Integration.Video.ExtractAudio.Config +instance GHC.Show.Show Integration.Video.ExtractAudio.ExtractionResult + +module Integration.Video.ExtractAudio.Internal +Runner :: (Text -> Task Error (Maybe Text)) -> (Int -> Text -> Array Text -> Path -> Task Error Completion) -> Runner +[run] :: Runner -> Int -> Text -> Array Text -> Path -> Task Error Completion +[which] :: Runner -> Text -> Task Error (Maybe Text) +buildFfmpegArgs :: Config -> Path -> Path -> Array Text +data Runner +executeExtraction :: (ToJSON command, KnownSymbol (NameOf command)) => Runner -> ActionContext -> Request command -> Task IntegrationError (Maybe CommandPayload) +extractAudio :: Runner -> Text -> FileAccessContext -> FileRef -> Config -> Task IntegrationError ExtractionResult +instance (Data.Aeson.Types.ToJSON.ToJSON command, GHC.TypeLits.KnownSymbol (Service.Command.Core.NameOf command)) => Integration.ToAction (Integration.Video.ExtractAudio.Request command) +realRunner :: Runner +resolveFfmpeg :: Runner -> Config -> Task IntegrationError Text diff --git a/core/core/Bytes.hs b/core/core/Bytes.hs index dbe033139..d4672c561 100644 --- a/core/core/Bytes.hs +++ b/core/core/Bytes.hs @@ -28,6 +28,7 @@ module Bytes ( concat, -- * Encoding toBase64, + fromBase64, ) where import Basics @@ -38,8 +39,12 @@ import Data.ByteString.Base64 qualified as Base64 import Data.ByteString.Lazy (LazyByteString) import Data.ByteString.Search qualified as ByteStringSearch import Data.Word (Word8) +import Result (Result) +import Result qualified import Task (Task) import Task qualified +import Text (Text) +import Text qualified import Prelude qualified as GhcPrelude @@ -223,4 +228,22 @@ concat bss = toBase64 :: Bytes -> Bytes toBase64 (INTERNAL_CORE_BYTES_CONSTRUCTOR bs) = Base64.encode bs - |> INTERNAL_CORE_BYTES_CONSTRUCTOR \ No newline at end of file + |> INTERNAL_CORE_BYTES_CONSTRUCTOR + + +-- | Decode Base64 back to the original bytes; the inverse of 'toBase64'. +-- +-- Decoding is strict: input must be padded standard Base64 with no +-- whitespace, so corrupted or truncated text is an 'Err', never a guess. +-- +-- >>> "aGVsbG8=" |> Text.toBytes |> Bytes.fromBase64 |> Result.map Text.fromBytes +-- Ok "hello" +-- +-- >>> "aGVsbG8" |> Text.toBytes |> Bytes.fromBase64 |> Result.isErr +-- True +fromBase64 :: Bytes -> Result Text Bytes +fromBase64 (INTERNAL_CORE_BYTES_CONSTRUCTOR bs) = + Base64.decode bs + |> Result.fromEither + |> Result.mapError Text.fromLinkedList + |> Result.map INTERNAL_CORE_BYTES_CONSTRUCTOR diff --git a/core/test/BytesSpec.hs b/core/test/BytesSpec.hs index b4bfc45a0..6fb86c8f3 100644 --- a/core/test/BytesSpec.hs +++ b/core/test/BytesSpec.hs @@ -3,7 +3,9 @@ module BytesSpec where import Bytes qualified import Core import LinkedList qualified +import Result qualified import Test +import Text qualified spec :: Spec Unit @@ -37,3 +39,46 @@ spec = parallel do -- The empty case is total, not an error. Bytes.empty |> Bytes.unpack |> shouldBe [] + + describe "fromBase64" do + it "round-trips every byte value through toBase64" \_ -> do + let allBytes = Bytes.pack [0 .. 255] + allBytes |> Bytes.toBase64 |> Bytes.fromBase64 |> shouldBe (Ok allBytes) + + -- Lengths 1, 2 and 3 cover both padding forms and no padding. + let prefixes = [1, 2, 3] |> LinkedList.map (\size -> Bytes.take size allBytes) + prefixes + |> LinkedList.map (\prefix -> prefix |> Bytes.toBase64 |> Bytes.fromBase64) + |> shouldBe (prefixes |> LinkedList.map Ok) + + it "round-trips random binary bytes through toBase64" \_ -> do + randomBytes <- Bytes.getRandom 1021 + randomBytes |> Bytes.toBase64 |> Bytes.fromBase64 |> shouldBe (Ok randomBytes) + + it "decodes the RFC 4648 known answers including padding" \_ -> do + let decodeText encoded = encoded |> Text.toBytes |> Bytes.fromBase64 + decodeText "Zg==" |> shouldBe (Ok (Text.toBytes "f")) + decodeText "Zm8=" |> shouldBe (Ok (Text.toBytes "fo")) + decodeText "Zm9v" |> shouldBe (Ok (Text.toBytes "foo")) + decodeText "Zm9vYg==" |> shouldBe (Ok (Text.toBytes "foob")) + decodeText "Zm9vYmE=" |> shouldBe (Ok (Text.toBytes "fooba")) + decodeText "Zm9vYmFy" |> shouldBe (Ok (Text.toBytes "foobar")) + + -- NUL, 0xFF and 0xFE exercise the "+" and "/" alphabet characters. + decodeText "AP/+" |> shouldBe (Ok (Bytes.pack [0x00, 0xFF, 0xFE])) + + it "decodes empty input to empty bytes" \_ -> do + Bytes.empty |> Bytes.fromBase64 |> shouldBe (Ok Bytes.empty) + + it "rejects malformed Base64 instead of guessing" \_ -> do + let decodeText encoded = encoded |> Text.toBytes |> Bytes.fromBase64 + -- Missing padding, extra padding, a character outside the alphabet, + -- embedded whitespace, the URL-safe alphabet, a truncated quantum, + -- and non-canonical trailing bits. + decodeText "Zm9vYg" |> shouldSatisfy Result.isErr + decodeText "Zg===" |> shouldSatisfy Result.isErr + decodeText "Zm9v!g==" |> shouldSatisfy Result.isErr + decodeText "Zm9v Yg==" |> shouldSatisfy Result.isErr + decodeText "AP_-" |> shouldSatisfy Result.isErr + decodeText "Z" |> shouldSatisfy Result.isErr + decodeText "Zh==" |> shouldSatisfy Result.isErr diff --git a/docs/changes/018-video-extract-audio.md b/docs/changes/018-video-extract-audio.md new file mode 100644 index 000000000..88ab6dbdd --- /dev/null +++ b/docs/changes/018-video-extract-audio.md @@ -0,0 +1,139 @@ +# Change 018: Add Integration.Video.ExtractAudio (ffmpeg audio extraction) + +Jess wants to transcribe uploaded videos. Upstream already turns audio into +text (`Integration.Audio.Transcribe`, ADR-0041) but cannot read a video file, +and no module calls `ffmpeg`. This change adds the one missing step: an +outbound integration that takes an uploaded video and hands back its audio +track as a mono 16 kHz WAV (`audio/wav`), which `Integration.Audio.Transcribe` +accepts. It mirrors `Integration.Pdf.ExtractText`: a pure request record for +Jess, and a `ToAction` instance in `Internal` that runs a local command-line +tool. Chaining the two integrations (video event → extract audio → upload the +audio → transcribe) is deliberately left to the application; one feature per +change. + +```yaml spec +issue: adhoc:video-extract-audio +kind: feature +touches: [outbound-integrations, core-primitives] +breaking: false +new-dependency: false # ffmpeg is a runtime tool, like pdftotext; no build-depends or flake input +new-capability: false +new-extension-point: false +``` + +## Contract delta + +```diff signatures ++ Bytes: fromBase64 :: Bytes -> Result Text Bytes ++ Integration.Video.ExtractAudio: data Config ++ Integration.Video.ExtractAudio: Config :: Int -> Int -> Int -> Int -> Maybe Text -> Config ++ Integration.Video.ExtractAudio: [timeoutSeconds] :: Config -> Int ++ Integration.Video.ExtractAudio: [sampleRateHz] :: Config -> Int ++ Integration.Video.ExtractAudio: [channels] :: Config -> Int ++ Integration.Video.ExtractAudio: [maxOutputBytes] :: Config -> Int ++ Integration.Video.ExtractAudio: [ffmpegPath] :: Config -> Maybe Text ++ Integration.Video.ExtractAudio: data ExtractionResult ++ Integration.Video.ExtractAudio: ExtractionResult :: Bytes -> Text -> ExtractionResult ++ Integration.Video.ExtractAudio: [audio] :: ExtractionResult -> Bytes ++ Integration.Video.ExtractAudio: [mimeType] :: ExtractionResult -> Text ++ Integration.Video.ExtractAudio: data Request command ++ Integration.Video.ExtractAudio: Request :: FileRef -> Config -> (ExtractionResult -> command) -> (Text -> command) -> Request command ++ Integration.Video.ExtractAudio: [fileRef] :: Request command -> FileRef ++ Integration.Video.ExtractAudio: [config] :: Request command -> Config ++ Integration.Video.ExtractAudio: [onSuccess] :: Request command -> ExtractionResult -> command ++ Integration.Video.ExtractAudio: [onError] :: Request command -> Text -> command ++ Integration.Video.ExtractAudio: defaultConfig :: Config ++ Integration.Video.ExtractAudio.Internal: data Runner ++ Integration.Video.ExtractAudio.Internal: Runner :: (Text -> Task Error (Maybe Text)) -> (Int -> Text -> Array Text -> Path -> Task Error Completion) -> Runner ++ Integration.Video.ExtractAudio.Internal: realRunner :: Runner ++ Integration.Video.ExtractAudio.Internal: [which] :: Runner -> Text -> Task Error (Maybe Text) ++ Integration.Video.ExtractAudio.Internal: [run] :: Runner -> Int -> Text -> Array Text -> Path -> Task Error Completion ++ Integration.Video.ExtractAudio.Internal: resolveFfmpeg :: Runner -> Config -> Task IntegrationError Text ++ Integration.Video.ExtractAudio.Internal: extractAudio :: Runner -> Text -> FileAccessContext -> FileRef -> Config -> Task IntegrationError ExtractionResult ++ Integration.Video.ExtractAudio.Internal: buildFfmpegArgs :: Config -> Path -> Path -> Array Text ++ Integration.Video.ExtractAudio.Internal: executeExtraction :: (ToJSON command, KnownSymbol (NameOf command)) => Runner -> ActionContext -> Request command -> Task IntegrationError (Maybe CommandPayload) ++ Integration.Video.ExtractAudio.Internal: instance (Data.Aeson.Types.ToJSON.ToJSON command, GHC.TypeLits.KnownSymbol (Service.Command.Core.NameOf command)) => Integration.ToAction (Integration.Video.ExtractAudio.Request command) +``` + +`Config` and `ExtractionResult` derive `Show`, `Eq` and `Generic`. They have no +JSON instances because `Bytes` has none; a command that must carry the audio +encodes it losslessly in its `onSuccess` callback with +`result.audio |> Bytes.toBase64 |> Text.fromBytes`. The module header shows +this, imports `Integration.Video.ExtractAudio.Internal ()` so the `ToAction` +instance is in scope, shows how the command handler decodes the Base64 text +back to the exact bytes (`audioBase64 |> Text.toBytes |> Bytes.fromBase64`), +and warns that the dispatcher's +`eventProcessingTimeoutMs` (default 30 s) must be raised above +`timeoutSeconds` (default 180 s), as `Integration.Audio.Transcribe` does. `defaultConfig` is 180 s, 16000 Hz, +1 channel, 25 000 000 bytes, `ffmpegPath = Nothing` (look up `ffmpeg` on the +`PATH`). The `Internal` signatures are for tests only; `Runner` is the +process boundary (`which` + `run`), and the `ToAction` instance always uses +`realRunner` (`Subprocess.which` / `Subprocess.runWithTimeout`). + +Supporting edits: `integrations/nhintegrations.cabal` (exposed modules, test +module), `.hlint.yaml` (adds `Integration.Video.ExtractAudio.Internal` to the +grandfathered `System.Directory` list, same reason and `belongs-in: Directory` +note as `Integration.Pdf.ExtractText.Internal`: it needs the OS temp +directory), and `docs/changes/test-surfaces.json` (see below). + +`Bytes.fromBase64` is the missing Core inverse of `Bytes.toBase64`, added +next to it in `core/core/Bytes.hs` so neither Jess nor the integration tests +reach for a raw `Data.ByteString.Base64` import. It reuses the +`base64-bytestring` package nhcore already depends on (no new +`build-depends` anywhere) and its strict decoder: unpadded, over-padded, +non-alphabet, whitespace, URL-safe and non-canonical input is an `Err`, so a +corrupted command fails closed instead of yielding different bytes. + +## Criteria + +The integrations hspec suite was not registered in `test-surfaces.json`. This +change registers it as `nhintegrations-test` with root `integrations/test`, so +the locators below resolve. + +All criteria are `unit` with boundary `none`: the process boundary is replaced +by a fake `Runner` that never starts a process. It records the bytes of the +input file it is given and, on exit code 0, writes a binary fake WAV to the +output path. The fixtures are binary (NUL bytes, every byte value, invalid +UTF-8), so an empty input file or a lossy text conversion fails. C8 runs the +real in-process `Service.Integration.Dispatcher` with an in-memory event store; +the fake runner's delay stands in for a slow video and its `TimeoutError` +stands in for `Subprocess.runWithTimeout` hitting `timeoutSeconds`, so the +millisecond timings keep the run under a few seconds. `subprocess:real` is NOT claimed, +because no test runs the real `ffmpeg`. CI and Jess's machines may not have +it, and the argument list and error handling are what this change owns. A real +`ffmpeg` run would need a registered `subprocess:real` fixture. That is a +possible follow-up. + +| ID | Behavior | Proving test | Level | Boundary | +|----|----------|--------------|-------|----------| +| C1 | ffmpeg missing: fails with a `ValidationError` that names ffmpeg and says to install it; nothing is run | `hspec:nhintegrations-test:integrations/test/Integration/Video/ExtractAudioSpec.hs#fails with an install hint when ffmpeg is missing` | unit | none | +| C2 | ffmpeg exits non-zero: the message reaches `onError`, and both temp files are removed | `hspec:nhintegrations-test:integrations/test/Integration/Video/ExtractAudioSpec.hs#reports the ffmpeg failure to onError and removes the temp files on a non-zero exit` | unit | none | +| C3 | Success: `onSuccess` gets the exact binary WAV bytes written at the output path and `audio/wav`; the temp files are removed; the exact binary video bytes were in the input file when ffmpeg ran | `hspec:nhintegrations-test:integrations/test/Integration/Video/ExtractAudioSpec.hs#returns the WAV bytes and audio/wav on success and removes the temp files`
`hspec:nhintegrations-test:integrations/test/Integration/Video/ExtractAudioSpec.hs#writes the video to the input file before running ffmpeg` | unit | none | +| C4 | ffmpeg arguments and timeout: defaults give `-vn -ac 1 -ar 16000 -fs 25000000` and 180 s; configured values, and `ffmpegPath`, are honored | `hspec:nhintegrations-test:integrations/test/Integration/Video/ExtractAudioSpec.hs#passes -vn -ac 1 -ar 16000 -fs 25000000 and the default 180 s timeout`
`hspec:nhintegrations-test:integrations/test/Integration/Video/ExtractAudioSpec.hs#passes the configured timeout, sample rate, channels and size cap`
`hspec:nhintegrations-test:integrations/test/Integration/Video/ExtractAudioSpec.hs#looks up and runs the configured ffmpegPath` | unit | none | +| C5 | `defaultConfig` is 180 s, 16000 Hz, mono, 25 MB, no explicit path | `hspec:nhintegrations-test:integrations/test/Integration/Video/ExtractAudioSpec.hs#defaults to 180 s, 16000 Hz, mono, 25 MB and ffmpeg from the PATH` | unit | none | +| C6 | File-access failures: uploads disabled fails the action; a missing file reaches `onError` without leaking the file reference | `hspec:nhintegrations-test:integrations/test/Integration/Video/ExtractAudioSpec.hs#fails when file uploads are not enabled`
`hspec:nhintegrations-test:integrations/test/Integration/Video/ExtractAudioSpec.hs#maps a missing file to a validation error without leaking the file reference` | unit | none | +| C7 | The documented command encoding is lossless: Base64 text decodes back to the exact binary WAV bytes | `hspec:nhintegrations-test:integrations/test/Integration/Video/ExtractAudioSpec.hs#round-trips binary WAV bytes through the Base64 command field` | unit | none | +| C8 | Dispatcher timeout interaction: with `eventProcessingTimeoutMs` above the extraction time, a slow extraction's `onSuccess` and an extraction timeout's `onError` reach the command endpoint; with a shorter dispatcher timeout, the worker is cancelled and neither arrives | `hspec:nhintegrations-test:integrations/test/Integration/Video/ExtractAudioSpec.hs#delivers onSuccess past a short deadline when the dispatcher timeout is raised above the extraction`
`hspec:nhintegrations-test:integrations/test/Integration/Video/ExtractAudioSpec.hs#cancels the extraction before onSuccess when the dispatcher timeout is shorter`
`hspec:nhintegrations-test:integrations/test/Integration/Video/ExtractAudioSpec.hs#reaches onError with the extraction timeout when the dispatcher timeout is longer`
`hspec:nhintegrations-test:integrations/test/Integration/Video/ExtractAudioSpec.hs#never reaches onError when the dispatcher timeout expires before the extraction timeout` | unit | none | +| C9 | `Bytes.fromBase64` inverts `Bytes.toBase64` for every byte value and random binary bytes, decodes independent RFC 4648 known answers with padding, decodes empty input to empty bytes, and rejects malformed input with an `Err` | `hspec:nhcore-test-core:core/test/BytesSpec.hs#round-trips every byte value through toBase64`
`hspec:nhcore-test-core:core/test/BytesSpec.hs#round-trips random binary bytes through toBase64`
`hspec:nhcore-test-core:core/test/BytesSpec.hs#decodes the RFC 4648 known answers including padding`
`hspec:nhcore-test-core:core/test/BytesSpec.hs#decodes empty input to empty bytes`
`hspec:nhcore-test-core:core/test/BytesSpec.hs#rejects malformed Base64 instead of guessing` | unit | none | + +## User impact + +None breaking. Jess gets a new integration, `Integration.Video.ExtractAudio`. +It needs `ffmpeg` on the `PATH` (or `ffmpegPath` set). If `ffmpeg` is missing, +the action fails with a message that tells her to install it. Failures in the +video itself (no audio track, unreadable file, timeout) reach her `onError` +command, provided the dispatcher's `eventProcessingTimeoutMs` is raised above +`timeoutSeconds` (the 30 s dispatcher default is shorter than the 180 s +extraction default; the module header shows the override). Output is capped +at 25 MB by default, so a very long video yields only its first part. A +command that carries the audio stores it as Base64 text and turns it back +into the exact audio with the new Core primitive `Bytes.fromBase64`, which +rejects malformed text instead of guessing. To transcribe a video, she chains this integration with +`Integration.Audio.Transcribe` using `mimeType = "audio/wav"`; the module +header shows how. No existing signature changes; `Bytes.fromBase64` is added. + +## ADR + +Not required — no trigger (breaking / new-dependency / new-capability / +new-extension-point all false). Related context: ADR-0041 (audio +transcription, two-persona model). diff --git a/docs/changes/test-surfaces.json b/docs/changes/test-surfaces.json index 32fa31a81..08eb0ae1f 100644 --- a/docs/changes/test-surfaces.json +++ b/docs/changes/test-surfaces.json @@ -30,6 +30,11 @@ "core/test-integration", "core/test" ] + }, + "nhintegrations-test": { + "roots": [ + "integrations/test" + ] } }, "boundaries": [ diff --git a/integrations/Integration/Video/ExtractAudio.hs b/integrations/Integration/Video/ExtractAudio.hs new file mode 100644 index 000000000..0588b79f6 --- /dev/null +++ b/integrations/Integration/Video/ExtractAudio.hs @@ -0,0 +1,257 @@ +-- | # Video Audio Extraction Integration +-- +-- This module extracts the audio track of an uploaded video file as a mono +-- 16 kHz WAV, using the local @ffmpeg@ command-line tool. It is the missing +-- first step of a video-transcription pipeline: "Integration.Audio.Transcribe" +-- accepts audio, so a video has to be turned into audio first. +-- +-- == Two-Persona Model +-- +-- Following ADR-0008, ADR-0015 and ADR-0041, this integration separates +-- concerns: +-- +-- * **Jess (Integration User)**: Configures extraction with pure records. +-- No @Task@, no subprocess handling, no temp files visible. +-- +-- * **Nick (Integration Developer)**: Implements 'ToAction' with ffmpeg +-- execution, temp file management, and error handling in +-- "Integration.Video.ExtractAudio.Internal". +-- +-- == Quick Start +-- +-- @ +-- import Bytes qualified +-- import Core +-- import Integration qualified +-- import Integration.Video.ExtractAudio qualified as VideoAudio +-- import Integration.Video.ExtractAudio.Internal () +-- import Text qualified +-- +-- lectureIntegrations :: Lecture -> LectureEvent -> Integration.Outbound +-- lectureIntegrations lecture event = case event of +-- VideoUploaded info -> Integration.batch +-- [ Integration.outbound VideoAudio.Request +-- { fileRef = info.fileRef +-- , config = VideoAudio.defaultConfig +-- , onSuccess = \\result -> RecordExtractedAudio +-- { lectureId = lecture.id +-- , audioBase64 = result.audio |> Bytes.toBase64 |> Text.fromBytes +-- , mimeType = result.mimeType +-- } +-- , onError = \\err -> AudioExtractionFailed +-- { lectureId = lecture.id +-- , error = err +-- } +-- } +-- ] +-- _ -> Integration.none +-- @ +-- +-- The @Internal ()@ import brings the 'Integration.ToAction' instance into +-- scope; without it @Integration.outbound@ does not accept a 'Request'. +-- +-- == Carrying the audio in a command +-- +-- Commands are serialized as JSON, and 'Bytes' has no JSON form. Encode +-- 'audio' as Base64 text (as above): every byte survives. Never use +-- @Text.fromBytes result.audio@ directly; WAV audio is not UTF-8 text. +-- +-- Your command handler decodes the text back to the exact WAV bytes before +-- uploading it, with 'Bytes.fromBase64'. Malformed text is an 'Err', so a +-- corrupted command never turns into silently wrong audio: +-- +-- @ +-- import Bytes qualified +-- import Core +-- import Text qualified +-- +-- decodeAudio :: Text -> Result Text Bytes +-- decodeAudio audioBase64 = +-- audioBase64 +-- |> Text.toBytes +-- |> Bytes.fromBase64 +-- @ +-- +-- Base64 makes the command about a third larger than the audio itself. +-- +-- == Dispatcher Timeout Warning +-- +-- The default 'timeoutSeconds' is 180 (3 minutes). The integration +-- dispatcher's @eventProcessingTimeoutMs@ defaults to 30000 (30 seconds). +-- __You MUST configure the dispatcher timeout to more than +-- @timeoutSeconds * 1000@ plus the time to fetch the video (or 'Nothing' to +-- disable it)__. Otherwise the dispatcher cancels the worker first: a slow +-- extraction never finishes, and an ffmpeg timeout never reaches 'onError'. +-- +-- @ +-- import Core +-- import Service.Application (Application) +-- import Service.Application qualified as Application +-- import Service.Integration.Dispatcher qualified as Dispatcher +-- +-- app :: Application +-- app = +-- Application.new +-- |> Application.withDispatcherConfig \@() (\\_ -> Dispatcher.defaultConfig +-- { Dispatcher.eventProcessingTimeoutMs = Just 200000 +-- }) +-- @ +-- +-- 'Service.Application.withDispatcherConfig' takes a function of your +-- application config; @\@()@ with @\\_ ->@ means the setting does not +-- depend on it. @Core@ provides 'Bytes', 'Text', 'Result', 'Just' and @|>@. +-- +-- If you raise 'timeoutSeconds', raise the dispatcher timeout with it. +-- +-- == Transcribing a video +-- +-- This integration only produces audio. To get a transcript, chain two +-- steps in your own domain (this module does not do the chaining for you): +-- +-- 1. A video event triggers 'Request'. Its 'onSuccess' command carries +-- the extracted 'audio'. +-- 2. Your command handler uploads that audio as a new file and records an +-- event with its file reference. +-- 3. That event triggers "Integration.Audio.Transcribe" with +-- @mimeType = "audio/wav"@: +-- +-- @ +-- Integration.outbound AudioTranscribe.Request +-- { fileRef = e.audioFileRef +-- , mimeType = "audio/wav" +-- , model = "google/gemini-2.5-flash" +-- , config = AudioTranscribe.defaultConfig +-- , onSuccess = \\transcript -> RecordTranscript { text = transcript.text } +-- , onError = \\err -> TranscriptionFailed { error = err } +-- } +-- @ +-- +-- == Output +-- +-- WAV, mono, 16 kHz by default ('mimeType' is @audio/wav@). The result is +-- capped at 'maxOutputBytes' (default 25 MB); ffmpeg stops writing at that +-- size, so a very long video yields only its first part. +-- +-- == Requirements +-- +-- This integration requires @ffmpeg@ on the @PATH@ (or set 'ffmpegPath'): +-- +-- @ +-- # Ubuntu/Debian +-- apt-get install ffmpeg +-- +-- # macOS +-- brew install ffmpeg +-- +-- # Nix +-- nix-shell -p ffmpeg +-- @ +-- +-- If @ffmpeg@ is missing, the integration fails with a validation error that +-- tells you to install it. +-- +-- == Known Limitations +-- +-- * A video without an audio track fails with an ffmpeg error. +-- * The whole video is held in memory and written to a temp file. +module Integration.Video.ExtractAudio + ( -- * Request Configuration (Jess's API) + Request (..) + , Config (..) + + -- * Result Types + , ExtractionResult (..) + + -- * Config Helpers + , defaultConfig + ) where + +import Basics +import Bytes (Bytes) +import Maybe (Maybe (..)) +import Service.FileUpload.Core (FileRef) +import Text (Text) + + +-- | Configuration for audio extraction. +-- +-- Use 'defaultConfig' for sensible defaults, then override specific fields: +-- +-- @ +-- VideoAudio.defaultConfig +-- { timeoutSeconds = 600 +-- , ffmpegPath = Just "/opt/ffmpeg/bin/ffmpeg" +-- } +-- @ +data Config = Config + { timeoutSeconds :: Int + -- ^ Timeout for the ffmpeg run (default: 180). The dispatcher's + -- @eventProcessingTimeoutMs@ must be longer; see the Dispatcher Timeout + -- Warning above. + , sampleRateHz :: Int + -- ^ Output sample rate in Hz (default: 16000) + , channels :: Int + -- ^ Output channel count (default: 1, mono) + , maxOutputBytes :: Int + -- ^ Stop writing audio after this many bytes (default: 25000000) + , ffmpegPath :: Maybe Text + -- ^ Explicit ffmpeg executable. Nothing means look up @ffmpeg@ on the PATH. + } + deriving (Show, Eq, Generic) + + +-- | Default configuration for audio extraction. +-- +-- * Timeout: 180 seconds +-- * Sample rate: 16000 Hz +-- * Channels: 1 (mono) +-- * Maximum output: 25 MB +-- * ffmpeg: found on the PATH +-- +-- >>> defaultConfig.timeoutSeconds +-- 180 +-- +-- >>> defaultConfig.sampleRateHz +-- 16000 +defaultConfig :: Config +defaultConfig = Config + { timeoutSeconds = 180 + , sampleRateHz = 16000 + , channels = 1 + , maxOutputBytes = 25000000 + , ffmpegPath = Nothing + } + + +-- | Result of audio extraction. +data ExtractionResult = ExtractionResult + { audio :: Bytes + -- ^ The extracted audio, as WAV + , mimeType :: Text + -- ^ Always @audio/wav@; pass it on to "Integration.Audio.Transcribe" + } + deriving (Show, Eq, Generic) + + +-- | The main request configuration record that Jess instantiates. +-- +-- The @command@ type parameter is the domain command emitted by +-- 'onSuccess' or 'onError' callbacks. +-- +-- == Fields +-- +-- * 'fileRef': Reference to the uploaded video file +-- * 'config': Extraction configuration (use 'defaultConfig') +-- * 'onSuccess': Callback that receives the audio and returns a domain command +-- * 'onError': Callback for error handling +data Request command = Request + { fileRef :: FileRef + -- ^ Reference to the uploaded video file + , config :: Config + -- ^ Extraction configuration + , onSuccess :: ExtractionResult -> command + -- ^ Callback for successful extraction + , onError :: Text -> command + -- ^ Callback for extraction errors + } + deriving (Generic) diff --git a/integrations/Integration/Video/ExtractAudio/Internal.hs b/integrations/Integration/Video/ExtractAudio/Internal.hs new file mode 100644 index 000000000..0cfbbbbe4 --- /dev/null +++ b/integrations/Integration/Video/ExtractAudio/Internal.hs @@ -0,0 +1,270 @@ +{-# LANGUAGE UndecidableInstances #-} + +-- | Internal implementation for video audio extraction. +-- +-- This module contains Nick's code - the 'ToAction' instance, ffmpeg +-- execution, and temp file handling. +-- +-- The process boundary is an explicit 'Runner' record. The 'ToAction' +-- instance uses 'realRunner'; tests pass a fake, so they never need ffmpeg. +-- +-- __This module is not exported to Jess.__ +module Integration.Video.ExtractAudio.Internal + ( -- * Process boundary + Runner (..) + , realRunner + + -- * For Testing Only + , executeExtraction + , extractAudio + , resolveFfmpeg + , buildFfmpegArgs + ) where + +import Array (Array) +import Array qualified +import Basics +import Bytes (Bytes) +import File qualified +import Integration qualified +import Integration.Video.ExtractAudio (Config (..), ExtractionResult (..), Request (..)) +import Json qualified +import Maybe (Maybe (..)) +import Maybe qualified +import Path (Path) +import Path qualified +import Result (Result (..)) +import Service.Command.Core (NameOf) +import Service.FileUpload.Core (FileAccessError (..), FileRef) +import Subprocess qualified +import System.Directory qualified as GhcDir +import Task (Task) +import Task qualified +import Text (Text) +import Text qualified +import Uuid qualified + + +-- | The process boundary: everything the extraction needs from the OS. +-- +-- Production code uses 'realRunner'. Tests substitute a fake that records +-- its arguments and writes a fake WAV file. +data Runner = Runner + { which :: Text -> Task Subprocess.Error (Maybe Text) + -- ^ Find an executable; Nothing when it is not installed + , run :: Int -> Text -> Array Text -> Path -> Task Subprocess.Error Subprocess.Completion + -- ^ Run an executable with a timeout (seconds), arguments and working directory + } + + +-- | The runner that talks to the real operating system. +realRunner :: Runner +realRunner = Runner + { which = Subprocess.which + , run = Subprocess.runWithTimeout + } + + +-- | The temp files of one extraction, all in one directory. +data Workspace = Workspace + { workDir :: Path + , inputPath :: Path + , outputPath :: Path + } + + +-- | ToAction instance that executes the audio extraction. +-- +-- This is the main entry point - when Jess writes: +-- +-- @ +-- Integration.outbound VideoAudio.Request { ... } +-- @ +-- +-- This instance converts the config into an executable action. +instance + (Json.ToJSON command, KnownSymbol (NameOf command)) => + Integration.ToAction (Request command) + where + toAction request = Integration.action \ctx -> do + executeExtraction realRunner ctx request + + +-- | Execute an audio extraction request. +-- +-- Setup problems (uploads disabled, ffmpeg not installed) fail the action. +-- Problems with the video itself reach Jess through 'onError'. +executeExtraction :: + forall command. + (Json.ToJSON command, KnownSymbol (NameOf command)) => + Runner -> + Integration.ActionContext -> + Request command -> + Task Integration.IntegrationError (Maybe Integration.CommandPayload) +executeExtraction runner ctx request = do + fileAccess <- requireFileAccess ctx + ffmpeg <- resolveFfmpeg runner request.config + outcome <- extractAudio runner ffmpeg fileAccess request.fileRef request.config + |> Task.asResult + case outcome of + Ok result -> + Integration.emitCommand (request.onSuccess result) + Err integrationError -> do + let errorText = integrationErrorToText integrationError + Integration.emitCommand (request.onError errorText) + + +-- | File uploads must be enabled, or there is nothing to read the video from. +requireFileAccess :: + Integration.ActionContext -> + Task Integration.IntegrationError Integration.FileAccessContext +requireFileAccess ctx = case ctx.fileAccess of + Nothing -> + Task.throw (Integration.ValidationError "File uploads not enabled. Cannot access video file.") + Just fileAccess -> + Task.yield fileAccess + + +-- | Find the ffmpeg executable, or fail with install instructions. +resolveFfmpeg :: Runner -> Config -> Task Integration.IntegrationError Text +resolveFfmpeg runner config = do + let requested = config.ffmpegPath |> Maybe.withDefault "ffmpeg" + found <- runner.which requested + |> Task.mapError subprocessToIntegrationError + case found of + Nothing -> + Task.throw (Integration.ValidationError [fmt|ffmpeg not found (looked for #{requested}). Please install ffmpeg (apt-get install ffmpeg, brew install ffmpeg) or set ffmpegPath.|]) + Just resolved -> + Task.yield resolved + + +-- | Fetch the video, run ffmpeg over it, and return the WAV audio. +-- +-- Both temp files are removed whether the run succeeds or fails. +extractAudio :: + Runner -> + Text -> + Integration.FileAccessContext -> + FileRef -> + Config -> + Task Integration.IntegrationError ExtractionResult +extractAudio runner ffmpeg fileAccess fileRef config = do + videoBytes <- fileAccess.retrieveFile fileRef + |> Task.mapError fileAccessToIntegrationError + workspace <- newWorkspace + let cleanup = do + File.deleteIfExists workspace.inputPath |> Task.ignoreError + File.deleteIfExists workspace.outputPath |> Task.ignoreError + convertToWav runner ffmpeg config workspace videoBytes + |> Task.finally cleanup + + +-- | Write the video, run ffmpeg over it, and read the WAV it produced. +convertToWav :: + Runner -> + Text -> + Config -> + Workspace -> + Bytes -> + Task Integration.IntegrationError ExtractionResult +convertToWav runner ffmpeg config workspace videoBytes = do + File.writeBytes workspace.inputPath videoBytes + |> Task.mapError (\_ -> Integration.UnexpectedError "Failed to write temp video file") + let args = buildFfmpegArgs config workspace.inputPath workspace.outputPath + completion <- runner.run config.timeoutSeconds ffmpeg args workspace.workDir + |> Task.mapError subprocessToIntegrationError + Task.unless (completion.exitCode == 0) do + let stderrText = Text.trim completion.stderr + Task.throw (Integration.PermanentFailure [fmt|ffmpeg failed: #{stderrText}|]) + audio <- File.readBytes workspace.outputPath + |> Task.mapError (\_ -> Integration.PermanentFailure "ffmpeg produced no audio output") + Task.yield ExtractionResult {audio, mimeType = "audio/wav"} + + +-- | Build the ffmpeg command arguments. +-- +-- Drops the video stream (@-vn@), converts to the configured channel count +-- and sample rate, and stops writing at the configured size (@-fs@). +-- +-- >>> import Array qualified +-- >>> import Integration.Video.ExtractAudio (defaultConfig) +-- >>> Array.toLinkedList (buildFfmpegArgs defaultConfig "in.mp4" "out.wav") +-- ["-nostdin","-v","error","-i","in.mp4","-vn","-ac","1","-ar","16000","-fs","25000000","out.wav"] +buildFfmpegArgs :: Config -> Path -> Path -> Array Text +buildFfmpegArgs config inputPath outputPath = do + let channelsText = Text.fromInt config.channels + let sampleRateText = Text.fromInt config.sampleRateHz + let maxBytesText = Text.fromInt config.maxOutputBytes + Array.fromLinkedList + [ "-nostdin" + , "-v", "error" + , "-i", Path.toText inputPath + , "-vn" + , "-ac", channelsText + , "-ar", sampleRateText + , "-fs", maxBytesText + , Path.toText outputPath + ] + + +-- | Pick unique input and output temp paths for one extraction. +newWorkspace :: Task Integration.IntegrationError Workspace +newWorkspace = do + tempDir <- GhcDir.getTemporaryDirectory + |> Task.fromIO + workDir <- case Path.fromLinkedList tempDir of + Nothing -> Task.throw (Integration.UnexpectedError "Invalid temp directory path") + Just dir -> Task.yield dir + tempId <- Uuid.generate + let tempIdText = Uuid.toText tempId + inputPath <- tempFilePath workDir [fmt|neohaskell-video-#{tempIdText}.input|] + outputPath <- tempFilePath workDir [fmt|neohaskell-video-#{tempIdText}.wav|] + Task.yield Workspace {workDir, inputPath, outputPath} + + +-- | A file name inside the temp directory. +tempFilePath :: Path -> Text -> Task Integration.IntegrationError Path +tempFilePath workDir fileName = case Path.fromText fileName of + Nothing -> Task.throw (Integration.UnexpectedError "Invalid temp file name") + Just name -> Task.yield (Path.append name workDir) + + +-- | Convert FileAccessError to IntegrationError. +fileAccessToIntegrationError :: FileAccessError -> Integration.IntegrationError +fileAccessToIntegrationError err = case err of + FileNotFound _ -> + Integration.ValidationError "File not found" + StateLookupFailed _ _ -> + Integration.UnexpectedError "Audio extraction failed: state lookup error" + NotOwner _ -> + Integration.AuthenticationError "Not authorized to access this file" + FileExpired _ -> + Integration.ValidationError "File has expired" + FileIsDeleted _ -> + Integration.ValidationError "File has been deleted" + BlobMissing _ -> + Integration.UnexpectedError "File blob is missing from storage" + StorageError _ -> + Integration.UnexpectedError "Audio extraction failed: storage error" + + +-- | Convert Subprocess.Error to IntegrationError. +subprocessToIntegrationError :: Subprocess.Error -> Integration.IntegrationError +subprocessToIntegrationError err = case err of + Subprocess.ProcessError msg -> + Integration.UnexpectedError [fmt|Process error: #{msg}|] + Subprocess.TimeoutError msg -> + Integration.PermanentFailure [fmt|Audio extraction timed out (#{msg}). The video may be too long.|] + Subprocess.ToolNotFound msg -> + Integration.ValidationError [fmt|Tool not found: #{msg}|] + + +-- | Convert IntegrationError to Text for the onError callback. +integrationErrorToText :: Integration.IntegrationError -> Text +integrationErrorToText err = case err of + Integration.ValidationError msg -> msg + Integration.AuthenticationError msg -> msg + Integration.NetworkError msg -> msg + Integration.RateLimited _ -> "Rate limited" + Integration.PermanentFailure msg -> msg + Integration.UnexpectedError msg -> msg diff --git a/integrations/nhintegrations.cabal b/integrations/nhintegrations.cabal index 37ade5aa6..8f329d958 100644 --- a/integrations/nhintegrations.cabal +++ b/integrations/nhintegrations.cabal @@ -100,6 +100,8 @@ library Integration.Pdf Integration.Pdf.ExtractText Integration.Pdf.ExtractText.Internal + Integration.Video.ExtractAudio + Integration.Video.ExtractAudio.Internal Integration.Brevo Integration.Brevo.Request Integration.Brevo.Response @@ -150,6 +152,7 @@ test-suite nhintegrations-test Integration.Acs.InternalSpec Integration.AzureAI.RequestSpec Integration.AzureAI.InternalSpec + Integration.Video.ExtractAudioSpec build-depends: hspec, nhintegrations diff --git a/integrations/test/Integration/Video/ExtractAudioSpec.hs b/integrations/test/Integration/Video/ExtractAudioSpec.hs new file mode 100644 index 000000000..b77a49530 --- /dev/null +++ b/integrations/test/Integration/Video/ExtractAudioSpec.hs @@ -0,0 +1,539 @@ +module Integration.Video.ExtractAudioSpec (spec) where + +import Array (Array) +import Array qualified +import AsyncTask qualified +import Auth.SecretStore.InMemory qualified as InMemorySecretStore +import Basics +import Bytes (Bytes) +import Bytes qualified +import ConcurrentMap (ConcurrentMap) +import ConcurrentMap qualified +import ConcurrentVar (ConcurrentVar) +import ConcurrentVar qualified +import DateTime qualified +import File qualified +import Integration (ActionContext (..), CommandPayload (..)) +import Integration (FileAccessContext (..)) +import Integration qualified +import Integration.Video.ExtractAudio (Config (..), ExtractionResult (..), Request (..)) +import Integration.Video.ExtractAudio qualified as VideoAudio +import Integration.Video.ExtractAudio.Internal (Runner (..), executeExtraction) +import Json qualified +import Lock (Lock) +import Map qualified +import Maybe (Maybe (..)) +import Path qualified +import Result (Result (..)) +import Service.Command.Core (NameOf) +import Service.Event (Event (..), StreamPosition (..)) +import Service.Event.EntityName (EntityName (..)) +import Service.Event.EventMetadata (EventMetadata (..)) +import Service.Event.StreamId qualified as StreamId +import Service.EventStore.InMemory qualified as InMemory +import Service.FileUpload.Core (FileAccessError (..), FileRef (..)) +import Service.Integration.DispatchRegistry qualified as DispatchRegistry +import Service.Integration.Dispatcher qualified as Dispatcher +import Service.Transport (EndpointHandler) +import Subprocess qualified +import Task (Task) +import Task qualified +import Test.Hspec +import Text (Text) +import Text qualified +import Uuid qualified +import Var (Var) +import Var qualified + + +spec :: Spec +spec = do + describe "Integration.Video.ExtractAudio.defaultConfig" do + it "defaults to 180 s, 16000 Hz, mono, 25 MB and ffmpeg from the PATH" do + VideoAudio.defaultConfig `shouldBe` Config + { timeoutSeconds = 180 + , sampleRateHz = 16000 + , channels = 1 + , maxOutputBytes = 25000000 + , ffmpegPath = Nothing + } + + describe "Integration.Video.ExtractAudio.executeExtraction" do + it "fails with an install hint when ffmpeg is missing" do + (outcome, calls) <- Task.runOrPanic (runScenario missingFfmpeg VideoAudio.defaultConfig) + case outcome of + Err (Integration.ValidationError message) -> do + Text.contains "ffmpeg" message `shouldBe` True + Text.contains "install" message `shouldBe` True + Err other -> + expectationFailure [fmt|expected a ValidationError, got #{other}|] + Ok _ -> + expectationFailure "expected an error when ffmpeg is missing" + Array.length calls `shouldBe` 0 + + it "reports the ffmpeg failure to onError and removes the temp files on a non-zero exit" do + (outcome, calls) <- Task.runOrPanic (runScenario (failingFfmpeg "Invalid data found") VideoAudio.defaultConfig) + case outcome of + Ok (Just payload) -> do + let encoded = Json.encodeText payload.commandData + Text.contains "ExtractionFailed" encoded `shouldBe` True + Text.contains "Invalid data found" encoded `shouldBe` True + _ -> + expectationFailure "expected the onError command" + Array.length calls `shouldBe` 1 + Task.runOrPanic (anyTempFileLeft calls) `shouldReturn` False + + it "returns the WAV bytes and audio/wav on success and removes the temp files" do + (outcome, calls) <- Task.runOrPanic (runScenario succeedingFfmpeg VideoAudio.defaultConfig) + case outcome of + Ok (Just payload) -> do + let encoded = Json.encodeText payload.commandData + Text.contains "AudioExtracted" encoded `shouldBe` True + Text.contains "audio/wav" encoded `shouldBe` True + decodedAudio payload `shouldBe` Ok fakeWavBytes + _ -> + expectationFailure "expected the onSuccess command" + Array.length calls `shouldBe` 1 + Task.runOrPanic (anyTempFileLeft calls) `shouldReturn` False + + it "writes the video to the input file before running ffmpeg" do + (_, calls) <- Task.runOrPanic (runScenario succeedingFfmpeg VideoAudio.defaultConfig) + let inputPresent = calls |> Array.map (\call -> call.callInputPresent) + inputPresent `shouldBe` Array.wrap True + let inputBytes = calls |> Array.map (\call -> call.callInputBytes) + inputBytes `shouldBe` Array.wrap (Just fakeVideoBytes) + + it "round-trips binary WAV bytes through the Base64 command field" do + let encodedAudio = fakeWavBytes |> Bytes.toBase64 |> Text.fromBytes + decodeAudio encodedAudio `shouldBe` Ok fakeWavBytes + Bytes.length fakeWavBytes `shouldBe` 268 + + it "passes -vn -ac 1 -ar 16000 -fs 25000000 and the default 180 s timeout" do + (_, calls) <- Task.runOrPanic (runScenario succeedingFfmpeg VideoAudio.defaultConfig) + case Array.get 0 calls of + Nothing -> expectationFailure "expected one ffmpeg call" + Just call -> do + let joined = Text.joinWith " " call.callArguments + Text.contains "-vn -ac 1 -ar 16000 -fs 25000000" joined `shouldBe` True + Text.contains "-nostdin" joined `shouldBe` True + call.callTimeout `shouldBe` 180 + call.callExecutable `shouldBe` "/usr/bin/ffmpeg" + + it "passes the configured timeout, sample rate, channels and size cap" do + let custom = VideoAudio.defaultConfig + { timeoutSeconds = 42 + , sampleRateHz = 8000 + , channels = 2 + , maxOutputBytes = 1000 + } + (_, calls) <- Task.runOrPanic (runScenario succeedingFfmpeg custom) + case Array.get 0 calls of + Nothing -> expectationFailure "expected one ffmpeg call" + Just call -> do + let joined = Text.joinWith " " call.callArguments + Text.contains "-vn -ac 2 -ar 8000 -fs 1000" joined `shouldBe` True + call.callTimeout `shouldBe` 42 + + it "looks up and runs the configured ffmpegPath" do + let custom = VideoAudio.defaultConfig {ffmpegPath = Just "/opt/ffmpeg/bin/ffmpeg"} + (_, calls) <- Task.runOrPanic (runScenario succeedingFfmpeg custom) + case Array.get 0 calls of + Nothing -> expectationFailure "expected one ffmpeg call" + Just call -> + call.callExecutable `shouldBe` "/opt/ffmpeg/bin/ffmpeg" + + it "fails when file uploads are not enabled" do + callsVar <- Task.runOrPanic (textTask (Var.new Array.empty)) + let runner = fakeRunner callsVar succeedingFfmpeg + ctx <- Task.runOrPanic (makeContext Nothing) + outcome <- Task.runOrPanic (textTask (executeExtraction runner ctx (makeRequest VideoAudio.defaultConfig) |> Task.asResult)) + case outcome of + Err (Integration.ValidationError message) -> + Text.contains "File uploads not enabled" message `shouldBe` True + _ -> + expectationFailure "expected a ValidationError for disabled file uploads" + + it "maps a missing file to a validation error without leaking the file reference" do + callsVar <- Task.runOrPanic (textTask (Var.new Array.empty)) + let runner = fakeRunner callsVar succeedingFfmpeg + let missingFile = FileAccessContext + { retrieveFile = \ref -> Task.throw (FileNotFound ref) + , getFileMetadata = \ref -> Task.throw (FileNotFound ref) + } + ctx <- Task.runOrPanic (makeContext (Just missingFile)) + outcome <- Task.runOrPanic (textTask (executeExtraction runner ctx (makeRequest VideoAudio.defaultConfig) |> Task.asResult)) + case outcome of + Ok (Just payload) -> do + let encoded = Json.encodeText payload.commandData + Text.contains "ExtractionFailed" encoded `shouldBe` True + Text.contains "File not found" encoded `shouldBe` True + Text.contains "00000000-0000-0000-0000-000000000001" encoded `shouldBe` False + _ -> + expectationFailure "expected the onError command" + calls <- Task.runOrPanic (textTask (Var.get callsVar)) + Array.length calls `shouldBe` 0 + + + describe "Integration.Video.ExtractAudio dispatcher timeout interaction" do + it "delivers onSuccess past a short deadline when the dispatcher timeout is raised above the extraction" do + let slowExtraction = succeedingFfmpeg {ffmpegDelayMs = slowExtractionMs} + (commands, calls) <- Task.runOrPanic (runThroughDispatcher slowExtraction (Just raisedDispatcherTimeoutMs)) + case commands |> Array.get 0 of + Just (AudioExtracted {audioBase64, mime}) -> do + decodeAudio audioBase64 `shouldBe` Ok fakeWavBytes + mime `shouldBe` "audio/wav" + _ -> + expectationFailure [fmt|expected the onSuccess command, got #{commands}|] + Array.length commands `shouldBe` 1 + Array.length calls `shouldBe` 1 + + it "cancels the extraction before onSuccess when the dispatcher timeout is shorter" do + let slowExtraction = succeedingFfmpeg {ffmpegDelayMs = slowExtractionMs} + (commands, calls) <- Task.runOrPanic (runThroughDispatcher slowExtraction (Just shortDispatcherTimeoutMs)) + commands `shouldBe` Array.empty + Array.length calls `shouldBe` 1 + + it "reaches onError with the extraction timeout when the dispatcher timeout is longer" do + let timingOut = succeedingFfmpeg {ffmpegDelayMs = slowExtractionMs, ffmpegTimesOut = True} + (commands, calls) <- Task.runOrPanic (runThroughDispatcher timingOut (Just raisedDispatcherTimeoutMs)) + case commands |> Array.get 0 of + Just (ExtractionFailed {reason}) -> + Text.contains "timed out" reason `shouldBe` True + _ -> + expectationFailure [fmt|expected the onError command, got #{commands}|] + Array.length commands `shouldBe` 1 + Array.length calls `shouldBe` 1 + + it "never reaches onError when the dispatcher timeout expires before the extraction timeout" do + let timingOut = succeedingFfmpeg {ffmpegDelayMs = slowExtractionMs, ffmpegTimesOut = True} + (commands, calls) <- Task.runOrPanic (runThroughDispatcher timingOut (Just shortDispatcherTimeoutMs)) + commands `shouldBe` Array.empty + Array.length calls `shouldBe` 1 + + +-- | How the fake ffmpeg behaves. +data Scenario = Scenario + { ffmpegInstalled :: Bool + , ffmpegExitCode :: Int + , ffmpegStderr :: Text + , ffmpegDelayMs :: Int + -- ^ Wall-clock time the fake run takes, standing in for a slow video + , ffmpegTimesOut :: Bool + -- ^ After the delay, fail as 'Subprocess.runWithTimeout' does when ffmpeg hits 'timeoutSeconds' + } + + +-- | One recorded invocation of the fake runner. +data Call = Call + { callTimeout :: Int + , callExecutable :: Text + , callArguments :: Array Text + , callInputPresent :: Bool + , callInputBytes :: Maybe Bytes + } + deriving (Eq, Show) + + +missingFfmpeg :: Scenario +missingFfmpeg = succeedingFfmpeg {ffmpegInstalled = False} + + +failingFfmpeg :: Text -> Scenario +failingFfmpeg stderrText = succeedingFfmpeg {ffmpegExitCode = 1, ffmpegStderr = stderrText} + + +succeedingFfmpeg :: Scenario +succeedingFfmpeg = Scenario + { ffmpegInstalled = True + , ffmpegExitCode = 0 + , ffmpegStderr = "" + , ffmpegDelayMs = 0 + , ffmpegTimesOut = False + } + + +-- | An MP4-style header followed by every byte value, so the fixture holds +-- NUL bytes and bytes that are not valid UTF-8. +fakeVideoBytes :: Bytes +fakeVideoBytes = + Bytes.append + (Bytes.pack [0x00, 0x00, 0x00, 0x18, 0x66, 0x74, 0x79, 0x70, 0x69, 0x73, 0x6F, 0x6D]) + (Bytes.pack [0 .. 255]) + + +-- | A RIFF/WAVE header followed by every byte value in descending order, so +-- the audio cannot survive a UTF-8 text round trip. +fakeWavBytes :: Bytes +fakeWavBytes = + Bytes.append + (Bytes.pack [0x52, 0x49, 0x46, 0x46, 0x24, 0x00, 0x00, 0x00, 0x57, 0x41, 0x56, 0x45]) + (Bytes.pack [255, 254 .. 0]) + + +-- | A runner that never starts a process. It records the bytes of the input +-- file it was given; after 'ffmpegDelayMs' it either fails with a timeout or, +-- on exit code 0, writes 'fakeWavBytes' to the output path, like ffmpeg would. +fakeRunner :: Var (Array Call) -> Scenario -> Runner +fakeRunner callsVar scenario = Runner + { which = \name -> + if not scenario.ffmpegInstalled + then Task.yield Nothing + else if Text.startsWith "/" name + then Task.yield (Just name) + else Task.yield (Just [fmt|/usr/bin/#{name}|]) + , run = \timeout executable arguments _ -> do + inputPresent <- case Array.get 4 arguments |> andThenPath of + Nothing -> Task.yield False + Just inputPath -> + File.exists inputPath + |> Task.mapError (\_ -> Subprocess.ProcessError "fake exists check failed") + inputBytes <- readInputBytes arguments + let call = Call + { callTimeout = timeout + , callExecutable = executable + , callArguments = arguments + , callInputPresent = inputPresent + , callInputBytes = inputBytes + } + recorded <- Var.get callsVar + Var.set (Array.push call recorded) callsVar + AsyncTask.sleep scenario.ffmpegDelayMs + Task.when scenario.ffmpegTimesOut do + Task.throw (Subprocess.TimeoutError [fmt|ffmpeg exceeded #{timeout} seconds|]) + Task.when (scenario.ffmpegExitCode == 0) do + case Array.last arguments |> andThenPath of + Nothing -> Task.throw (Subprocess.ProcessError "fake runner: no output path") + Just outputPath -> + File.writeBytes outputPath fakeWavBytes + |> Task.mapError (\_ -> Subprocess.ProcessError "fake runner: write failed") + Task.yield Subprocess.Completion + { exitCode = scenario.ffmpegExitCode + , stdout = "" + , stderr = scenario.ffmpegStderr + } + } + + +-- | The bytes ffmpeg would read from its @-i@ input file; Nothing when unreadable. +readInputBytes :: Array Text -> Task Subprocess.Error (Maybe Bytes) +readInputBytes arguments = case Array.get 4 arguments |> andThenPath of + Nothing -> Task.yield Nothing + Just inputPath -> + File.readBytes inputPath + |> Task.map Just + |> Task.recover (\_ -> Task.yield Nothing) + + +andThenPath :: Maybe Text -> Maybe Path.Path +andThenPath maybeText = case maybeText of + Nothing -> Nothing + Just text -> Path.fromText text + + +-- | The input and output temp paths ffmpeg was given. +tempPathsOf :: Call -> Array Path.Path +tempPathsOf call = do + let inputPath = Array.get 4 call.callArguments |> andThenPath + let outputPath = Array.last call.callArguments |> andThenPath + case (inputPath, outputPath) of + (Just input, Just output) -> Array.fromLinkedList [input, output] + _ -> Array.empty + + +-- | Pins the error type of a helper task to 'Text' so 'Task.runOrPanic' can show it. +textTask :: Task Text value -> Task Text value +textTask task = task + + +anyTempFileLeft :: Array Call -> Task Text Bool +anyTempFileLeft calls = do + flags <- calls + |> Array.flatMap tempPathsOf + |> Task.mapArray (\path -> File.exists path |> Task.recover (\_ -> Task.yield False)) + Task.yield (Array.any (\present -> present) flags) + + +makeContext :: Maybe Integration.FileAccessContext -> Task Text ActionContext +makeContext fileAccess = do + stubStore <- InMemorySecretStore.new + emptyLocks <- ConcurrentMap.new :: Task Text (ConcurrentMap Text Lock) + Task.yield ActionContext + { secretStore = stubStore + , providerRegistry = Integration.fromMap Map.empty + , refreshLocks = emptyLocks + , fileAccess = fileAccess + , outboundDispatch = DispatchRegistry.empty + } + + +runScenario :: + Scenario -> + Config -> + Task Text (Result Integration.IntegrationError (Maybe CommandPayload), Array Call) +runScenario scenario config = do + callsVar <- Var.new Array.empty + let runner = fakeRunner callsVar scenario + ctx <- makeContext (Just videoAccess) + outcome <- executeExtraction runner ctx (makeRequest config) |> Task.asResult + calls <- Var.get callsVar + Task.yield (outcome, calls) + + +-- | Uploads that always hand back 'fakeVideoBytes'. +videoAccess :: FileAccessContext +videoAccess = FileAccessContext + { retrieveFile = \_ -> Task.yield fakeVideoBytes + , getFileMetadata = \_ -> Task.throw (StorageError "metadata not available") + } + + +makeRequest :: Config -> Request TestCommand +makeRequest config = Request + { fileRef = FileRef "00000000-0000-0000-0000-000000000001" + , config = config + , onSuccess = \result -> AudioExtracted + { audioBase64 = result.audio |> Bytes.toBase64 |> Text.fromBytes + , mime = result.mimeType + } + , onError = \reason -> ExtractionFailed {reason} + } + + +-- | The decoding step the module documentation tells Jess to use. +decodeAudio :: Text -> Result Text Bytes +decodeAudio audioBase64 = + audioBase64 + |> Text.toBytes + |> Bytes.fromBase64 + + +-- | The audio carried by an emitted onSuccess command, decoded from JSON and Base64. +decodedAudio :: CommandPayload -> Result Text Bytes +decodedAudio payload = + case Json.decode payload.commandData of + Ok (AudioExtracted {audioBase64}) -> decodeAudio audioBase64 + Ok other -> Err [fmt|expected AudioExtracted, got #{other}|] + Err decodeError -> Err decodeError + + +-- | Longer than 'shortDispatcherTimeoutMs', far shorter than 'raisedDispatcherTimeoutMs'. +slowExtractionMs :: Int +slowExtractionMs = 400 + + +shortDispatcherTimeoutMs :: Int +shortDispatcherTimeoutMs = 150 + + +raisedDispatcherTimeoutMs :: Int +raisedDispatcherTimeoutMs = 3000 + + +-- | Longest wait for the dispatcher to deliver a command. A command that has +-- not arrived by then was cancelled; the extraction itself ends long before. +commandWaitMs :: Int +commandWaitMs = 1500 + + +-- | Run one extraction the way an application does: a dispatcher worker +-- processes a video event under the given 'eventProcessingTimeoutMs' and +-- sends the emitted command to its endpoint. Returns the commands received +-- and the ffmpeg runs that started. +runThroughDispatcher :: Scenario -> Maybe Int -> Task Text (Array TestCommand, Array Call) +runThroughDispatcher scenario dispatcherTimeoutMs = do + callsVar <- Var.new Array.empty + received <- ConcurrentVar.containing Array.empty + context <- makeContext (Just videoAccess) + eventStore <- InMemory.new |> Task.mapError (\err -> [fmt|#{err}|]) + dispatcher <- + Dispatcher.newWithLifecycleConfig + (dispatcherConfig dispatcherTimeoutMs) + eventStore + (Array.wrap (extractionOutboundRunner (fakeRunner callsVar scenario))) + Array.empty + (Map.empty |> Map.set "TestCommand" (recordingEndpoint received)) + context + event <- videoUploadedEvent + Dispatcher.dispatch dispatcher event + waitForCommand received commandWaitMs + Dispatcher.shutdown dispatcher + commands <- ConcurrentVar.peek received + calls <- Var.get callsVar + Task.yield (commands, calls) + + +-- | Dispatcher settings for a deterministic test: no reaper, no retries. +dispatcherConfig :: Maybe Int -> Dispatcher.DispatcherConfig +dispatcherConfig eventProcessingTimeoutMs = + Dispatcher.defaultConfig + { Dispatcher.enableReaper = False + , Dispatcher.eventProcessingTimeoutMs = eventProcessingTimeoutMs + , Dispatcher.maxEventRetries = 0 + } + + +-- | The outbound runner an application derives for 'Request', with the fake process boundary. +extractionOutboundRunner :: Runner -> Dispatcher.OutboundRunner +extractionOutboundRunner runner = Dispatcher.OutboundRunner + { entityTypeName = "Lecture" + , processEvent = \ctx _eventStore _event -> do + emitted <- executeExtraction runner ctx (makeRequest VideoAudio.defaultConfig) + |> Task.mapError (\err -> [fmt|#{err}|]) + case emitted of + Just payload -> Task.yield (Array.wrap payload) + Nothing -> Task.yield Array.empty + } + + +-- | A command endpoint that decodes and keeps every command it receives. +recordingEndpoint :: ConcurrentVar (Array TestCommand) -> EndpointHandler +recordingEndpoint received _requestContext commandBytes _respond = + case Json.decodeBytes commandBytes of + Ok command -> received |> ConcurrentVar.modify (Array.push command) + Err decodeError -> Task.throw [fmt|undecodable command: #{decodeError}|] + + +-- | Poll until a command arrives or the wait runs out. +waitForCommand :: ConcurrentVar (Array TestCommand) -> Int -> Task Text Unit +waitForCommand received remainingMs = do + commands <- ConcurrentVar.peek received + if Array.length commands > 0 || remainingMs <= 0 + then Task.yield unit + else do + AsyncTask.sleep 25 + waitForCommand received (remainingMs - 25) + + +videoUploadedEvent :: Task Text (Event Json.Value) +videoUploadedEvent = do + now <- DateTime.now + Task.yield Event + { entityName = EntityName "Lecture" + , streamId = StreamId.fromTextUnsafe "lecture-1" + , event = Json.encode ("VideoUploaded" :: Text) + , metadata = EventMetadata + { eventId = Uuid.nil + , relatedUserSub = Nothing + , correlationId = Nothing + , causationId = Nothing + , createdAt = now + , localPosition = Just (StreamPosition 1) + , globalPosition = Just (StreamPosition 1) + } + } + + +-- | Minimal command type satisfying the ToAction constraints. +data TestCommand + = AudioExtracted {audioBase64 :: Text, mime :: Text} + | ExtractionFailed {reason :: Text} + deriving (Eq, Show, Generic) + + +instance Json.ToJSON TestCommand + + +instance Json.FromJSON TestCommand + + +type instance NameOf TestCommand = "TestCommand"