broken transcode

5 files changed, 92 insertions, 44 deletions

diff --git a/base/src/cache.rs b/base/src/cache.rs
index 7705a14..ee13b73 100644
--- a/base/src/cache.rs
+++ b/base/src/cache.rs
@@ -3,7 +3,7 @@ use anyhow::{anyhow, Context};
 use base64::Engine;
 use bincode::{Decode, Encode};
 use jellycommon::AssetLocation;
-use log::info;
+use log::{info, warn};
 use std::{
     any::Any,
     collections::{BTreeMap, HashMap},
@@ -54,7 +54,14 @@ where
         let f = tokio::fs::File::create(location.path())
             .await
             .context("creating new cache file")?;
-        generate(f).await?;
+        match generate(f).await {
+            Ok(()) => (),
+            Err(e) => {
+                warn!("cache generation failed, unlinking entry");
+                tokio::fs::remove_file(location.path()).await?;
+                return Err(e);
+            }
+        }
     }
     drop(_guard);
     Ok(location)
@@ -70,7 +77,14 @@ where
     let exists = location.path().exists();
     if !exists {
         let f = std::fs::File::create(location.path()).context("creating new cache file")?;
-        generate(f)?;
+        match generate(f) {
+            Ok(()) => (),
+            Err(e) => {
+                warn!("cache generation failed, unlinking entry");
+                std::fs::remove_file(location.path())?;
+                return Err(e);
+            }
+        }
     }
     drop(_guard);
     Ok(location)
diff --git a/common/src/stream.rs b/common/src/stream.rs
index e6fe767..1fb3761 100644
--- a/common/src/stream.rs
+++ b/common/src/stream.rs
@@ -13,8 +13,7 @@ pub struct StreamSpec {
     pub tracks: Vec<usize>,
     pub format: StreamFormat,
     pub webm: Option<bool>,
-    pub abr: Option<usize>,
-    pub vbr: Option<usize>,
+    pub bitrate: Option<usize>,
     pub width: Option<usize>,
     pub index: Option<usize>,
 }
@@ -39,8 +38,7 @@ impl Default for StreamSpec {
             format: StreamFormat::Matroska,
             webm: Some(true),
             width: None,
-            abr: None,
-            vbr: None,
+            bitrate: None,
             index: None,
         }
     }
@@ -64,11 +62,8 @@ impl StreamSpec {
             )
             .unwrap();
         }
-        if let Some(abr) = self.abr {
-            write!(u, "&abr={abr}").unwrap();
-        }
-        if let Some(vbr) = self.vbr {
-            write!(u, "&vbr={vbr}").unwrap();
+        if let Some(bitrate) = self.bitrate {
+            write!(u, "&bitrate={bitrate}").unwrap();
         }
         if let Some(index) = self.index {
             write!(u, "&index={index}").unwrap();
diff --git a/stream/src/segment.rs b/stream/src/segment.rs
index 78afc4f..02f3100 100644
--- a/stream/src/segment.rs
+++ b/stream/src/segment.rs
@@ -4,39 +4,69 @@
     Copyright (C) 2023 metamuffin <metamuffin.org>
 */
 use anyhow::{anyhow, bail, Result};
-use jellybase::CONF;
+use jellybase::{AssetLocationExt, CONF};
 use jellycommon::{stream::StreamSpec, LocalTrack, Node};
+use jellytranscoder::snippet::{transcode, Encoding};
 use log::warn;
-use tokio::io::DuplexStream;
+use tokio::{fs::File, io::DuplexStream};
 use tokio_util::io::SyncIoBridge;
 
 pub async fn segment_stream(
     node: Node,
     track_sources: Vec<LocalTrack>,
     spec: StreamSpec,
-    b: DuplexStream,
+    mut b: DuplexStream,
 ) -> Result<()> {
-    let b = SyncIoBridge::new(b);
-
     if spec.tracks.len() != 1 {
         bail!("unsupported number of tracks for segment, must be exactly one");
     }
     let track = spec.tracks[0];
     let n = spec.index.ok_or(anyhow!("segment index missing"))?;
 
-    tokio::task::spawn_blocking(move || {
-        if let Err(err) = jellyremuxer::write_snippet_into(
-            b,
-            &CONF.library_path,
-            &node.public,
-            track_sources,
-            track,
-            spec.webm.unwrap_or(false),
-            n,
-        ) {
-            warn!("segment stream error: {err}");
-        }
-    });
+    if let Some(width) = spec.width {
+        let location = transcode(
+            "",
+            Encoding::Video {
+                codec: "libsvtav1",
+                preset: 8,
+                bitrate: spec.bitrate.unwrap_or(2_000_000),
+                width,
+            },
+            move |b| {
+                tokio::task::spawn_blocking(move || {
+                    if let Err(err) = jellyremuxer::write_snippet_into(
+                        SyncIoBridge::new(b),
+                        &CONF.library_path,
+                        &node.public,
+                        track_sources,
+                        track,
+                        false,
+                        n,
+                    ) {
+                        warn!("segment stream error: {err}");
+                    }
+                });
+            },
+        )
+        .await?;
+        let mut output = File::open(location.path()).await?;
+        tokio::io::copy(&mut output, &mut b).await?;
+    } else {
+        let b = SyncIoBridge::new(b);
+        tokio::task::spawn_blocking(move || {
+            if let Err(err) = jellyremuxer::write_snippet_into(
+                b,
+                &CONF.library_path,
+                &node.public,
+                track_sources,
+                track,
+                spec.webm.unwrap_or(false),
+                n,
+            ) {
+                warn!("segment stream error: {err}");
+            }
+        });
+    }
 
     Ok(())
 }
diff --git a/transcoder/src/lib.rs b/transcoder/src/lib.rs
index 167303d..240d37a 100644
--- a/transcoder/src/lib.rs
+++ b/transcoder/src/lib.rs
@@ -4,5 +4,6 @@
     Copyright (C) 2023 metamuffin <metamuffin.org>
 */
 #![feature(async_closure)]
+#![feature(exit_status_error)]
 pub mod image;
 pub mod snippet;
diff --git a/transcoder/src/snippet.rs b/transcoder/src/snippet.rs
index 56bd523..5afeb15 100644
--- a/transcoder/src/snippet.rs
+++ b/transcoder/src/snippet.rs
@@ -4,10 +4,14 @@
     Copyright (C) 2023 metamuffin <metamuffin.org>
 */
 
-use jellybase::{cache::async_cache_file, AssetLocationExt};
+use jellybase::cache::async_cache_file;
 use jellycommon::AssetLocation;
+use log::info;
 use std::process::Stdio;
-use tokio::{fs::File, process::Command};
+use tokio::{
+    io::copy,
+    process::{ChildStdin, Command},
+};
 
 #[derive(Debug)]
 pub enum Encoding {
@@ -19,16 +23,15 @@ pub enum Encoding {
     },
 }
 
-pub async fn transcode(asset: AssetLocation, enc: Encoding) -> anyhow::Result<AssetLocation> {
-    let original_path = asset.path();
-    let asset = asset.clone();
+pub async fn transcode(
+    key: &str,
+    enc: Encoding,
+    input: impl FnOnce(ChildStdin),
+) -> anyhow::Result<AssetLocation> {
     Ok(async_cache_file(
-        &[
-            "snip-tc",
-            original_path.as_os_str().to_str().unwrap(),
-            &format!("{enc:?}"),
-        ],
-        move |output| async move {
+        &["snip-tc", key, &format!("{enc:?}")],
+        move |mut output| async move {
+            info!("transcoding snippet {key}");
             let args = match enc {
                 Encoding::Video {
                     codec,
@@ -50,14 +53,19 @@ pub async fn transcode(asset: AssetLocation, enc: Encoding) -> anyhow::Result<As
             let mut proc = Command::new("ffmpeg")
                 .stdin(Stdio::piped())
                 .stdout(Stdio::piped())
-                .args(&["-f", "mkv", "-i", "pipe:0"])
+                .args(&["-f", "matroska", "-i", "pipe:0"])
                 .args(args)
-                .args(&["-f", "mkv", "pipe:1"])
+                .args(&["-f", "webm", "pipe:1"])
                 .spawn()?;
 
-            let mut stdin = proc.stdin.take().unwrap();
+            let stdin = proc.stdin.take().unwrap();
             let mut stdout = proc.stdout.take().unwrap();
 
+            input(stdin);
+            copy(&mut stdout, &mut output).await?;
+
+            proc.wait().await.unwrap().exit_ok()?;
+            info!("done");
             Ok(())
         },
     )