console.log(event);

      if ('Chunk' in m) {
        console.log(m);
        out[m.Chunk.channel].push(ansi_up.ansi_to_html(m.Chunk.content));
      } else if ('Status' in m) {
        console.log(m);
        ended = m.Status.ended;
        status = m.Status.status;

      if (typeof m == 'object') {
        if (m?.Chunk) {
          console.log(m);
          out[m.Chunk.channel].push(ansi_up.ansi_to_html(m.Chunk.content));
          if (m.Chunk.channel == 0 && out[1].reduce((a, b) => a + b.length, 0)) {
            document.getElementById('stdout-bottom')?.scrollIntoView();
          } else {
            document.getElementById('stderr-bottom')?.scrollIntoView();
          }
        } else if (m?.Status) {
          console.log(m);
          ended = m.Status.ended;
          status = m.Status.status;
        }

    socket.onopen = () => {
      // socket.send('{ "State": { "stdout": 0, "stderr": 0 } }');
    };

    <div id="stdout-bottom"></div>

    <div id="stderr-bottom"></div>

  import { server } from '../../../helpers';
  import { onMount } from 'svelte';

  let jobs = $derived(data.jobs);
  let socket: WebSocket | undefined = undefined;
  function websocket() {
    socket = new WebSocket(`${server}/api/job/${params.user}/${params.repo}/ws`);
    socket.onmessage = (event) => {
      const j = JSON.parse(event.data);
      if (typeof j == 'object') {
        if (j?.jobs) {
          jobs = j.jobs;
        }
      }
    };
    socket.onclose = () => {
      socket = undefined;
    };
  }
  onMount(() => {
    setInterval(() => {
      if (!socket) websocket();
    }, 1000);
  });

      <thead> <tr><td>Job</td><td>Started</td><td>Ended</td><td>Status</td> </tr></thead>

      <thead><tr><td>Job</td><td>Started</td><td>Ended</td><td>Status</td><td></td></tr></thead>

        {#each data.jobs as job (job.id)}

        {#each jobs as job (job.id)}

            </td>
            <td>
              <div class="flex flex-wrap gap-3">
                <form action="/api/job/{params.user}/{params.repo}/{job.id}/retry" method="post">
                  <button class="btn btn-primary btn-sm" disabled={!job.ended}>Retry</button>
                </form>
                <form action="/api/job/{params.user}/{params.repo}/{job.id}/kill" method="post">
                  <button class="btn btn-error btn-sm" disabled={job.ended}>Kill</button>
                </form>
              </div>

  console.log('handle', event);

    console.log('setCookie', setCookie);

        hash: libpijul::pristine::Hash,

        hash: pijul_core::pristine::Hash,

        hash: libpijul::pristine::Hash,

        hash: pijul_core::pristine::Hash,

        hash: libpijul::pristine::Hash,

        hash: pijul_core::pristine::Hash,

        p.push(libpijul::DOT_DIR);

        p.push(pijul_core::DOT_DIR);

        hash: libpijul::Hash,

        hash: pijul_core::Hash,

        libpijul::changestore::filesystem::push_filename(&mut path, &hash);

        pijul_core::changestore::filesystem::push_filename(&mut path, &hash);

    ChangeNotFound { repo: uuid::Uuid, hash: libpijul::Hash },

    ChangeNotFound { repo: uuid::Uuid, hash: pijul_core::Hash },

futures = "*"
tokio = { version = "1.48", features = [ "net", "time", "rt-multi-thread", "macros", "io-util", "fs", "sync", "signal" ] }
memmap = "*"
log = "*"
env_logger = "*"
serde_derive = "*"
serde = "*"
rand = "0.9"
uuid = { version = "*", features = [ "v4", "serde" ] }
toml = "*"
postgres-types = { version = "*", features = [ "derive" ] }
tokio-postgres = { version = "*", features = [ "with-uuid-1", "with-chrono-0_4" ] }
postgres-openssl = "*"
libpijul = { version = "1.0.0-beta.10", features = [ "tarball" ] }
clap = "~4.5.53"
regex = "*"
lazy_static = "*"
libc = "*"
bytes = "1.11"
deadpool-postgres = "*"
bincode = "*"
thiserror = "*"
webpki = "*"
openssl = "*"
tokio-openssl = "*"
byteorder = "*"
indexmap = "*"

futures.workspace = true
tokio.workspace = true
serde_derive.workspace = true
uuid.workspace = true
pijul-core.workspace = true
thiserror.workspace = true
serde.workspace = true

ALTER TABLE jobs ADD COLUMN repo_state TEXT, ADD COLUMN channel TEXT NOT NULL DEFAULT 'main', ADD COLUMN script TEXT;

ALTER TABLE jobs DROP COLUMN repo_state, DROP COLUMN channel, DROP COLUMN script;

      zstd xxHash

      zstd xxhash

      rustfmt

      # rustfmt
      dbus
      nodejs
      typescript-language-server

use ci::Message;

use axum::{
    Json, Router,
    extract::{Path, State},
    http::StatusCode,
    response::{IntoResponse, Response},
    routing::{get, post},
};
// use ci::Message;
use clap::Parser;

use std::io::Read;
use std::net::{Ipv6Addr, SocketAddr, ToSocketAddrs};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime};

// use std::io::Read;
use axum_extra::TypedHeader;
use axum_extra::headers::Range;
use ci::*;
use pijul_core::MutTxnT;
use std::collections::HashMap;
use std::net::{Ipv6Addr, ToSocketAddrs};
use std::path::PathBuf;
use std::process::Stdio;
use std::sync::Arc;
use tempfile;
use thiserror::*;
use tokio::fs::OpenOptions;

    key_path: String,

    timeout_secs: usize,
    server_public_keys: Vec<String>,
    log_path: String,
    tarball_path: String,
}
#[derive(Serialize, Deserialize)]
struct BuildResult {
    finished: chrono::DateTime<chrono::Utc>,
    status: Option<i32>,
    link: Option<PathBuf>,
    job: ci::Job,

    nest_url: String,
    repo_path: PathBuf,
    ci_path: PathBuf,

use clap::*;

#[derive(Clone)]
pub struct Config {
    config: Arc<ConfigFile>,
    jobs: Arc<
        std::sync::Mutex<
            std::collections::HashMap<
                uuid::Uuid,
                (
                    tokio::sync::oneshot::Sender<()>,
                    tokio::sync::watch::Receiver<std::option::Option<i32>>,
                ),
            >,
        >,
    >,
}

    pijul_interaction::set_context(pijul_interaction::InteractiveContext::NotInteractive);

    let config = Arc::new(thrussh::client::Config::default());

    let key = Arc::new(thrussh_keys::load_secret_key(&conf.key_path, None).unwrap());
    let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
    let client = CiClient {
        process: Arc::new(Mutex::new(Process::default())),
        log_path: Path::new(&conf.log_path).to_path_buf(),
        tarball_path: Path::new(&conf.tarball_path).to_path_buf(),
        last_window_adjustment: SystemTime::now(),
        server_public_keys: Arc::new(
            conf.server_public_keys
                .iter()
                .map(|p| thrussh_keys::parse_public_key_base64(p).unwrap())
                .collect(),
        ),
        sender,

    let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
    let config = Config {
        config: Arc::new(conf),
        jobs: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),

    loop {
        if let Err(e) = client
            .protocol(&addr, config.clone(), key.clone(), &mut receiver)
            .await
        {
            error!("restarting because of error: {:?}", e)
        }
        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
    }

    let app = Router::new()
        .route("/trigger", post(trigger))
        .route("/stdout/{job}", get(stdout))
        .route("/stderr/{job}", get(stderr))
        .route("/status/{job}", get(status))
        .with_state(config.clone());
    axum::serve(listener, app).await.unwrap();

#[derive(Clone, Debug)]
pub struct CiClient {
    process: Arc<Mutex<Process>>,
    tarball_path: PathBuf,
    log_path: PathBuf,
    last_window_adjustment: SystemTime,
    server_public_keys: Arc<Vec<thrussh_keys::key::PublicKey>>,
    sender: tokio::sync::mpsc::Sender<(ci::Job, Option<i32>, Option<PathBuf>)>,

async fn status(
    State(config): State<Config>,
    Path(job): Path<uuid::Uuid>,
) -> Result<Json<Status>, Error> {
    let s =
        tokio::fs::read_to_string(&config.config.ci_path.join(&format!("{}.status", job))).await?;
    Ok(Json(serde_json::from_str(&s)?))

#[derive(Debug, Default)]
struct Process {
    child: Option<tokio::task::JoinHandle<Result<(), anyhow::Error>>>,
    job: Option<ci::Job>,
    tarball: Option<(std::fs::File, usize)>,
}

use axum_range::KnownSize;
use axum_range::Ranged;

impl Process {
    fn is_ready(&self) -> bool {
        self.child.is_none() && self.job.is_none()

async fn stdout(
    State(config): State<Config>,
    Path(job): Path<uuid::Uuid>,
    range: Option<TypedHeader<Range>>,
) -> Response {
    debug!("stdout {:?} {:?}", job, range);
    let Ok(mut file) =
        tokio::fs::File::open(&config.config.ci_path.join(&format!("{}.stdout", job))).await
    else {
        return StatusCode::NOT_FOUND.into_response();
    };
    if let Some(TypedHeader(range)) = range {
        let body = KnownSize::file(file).await.unwrap();
        Ranged::new(Some(range), body).into_response()
    } else {
        use tokio::io::AsyncReadExt;
        let mut v = Vec::new();
        file.read_to_end(&mut v).await.unwrap();
        v.into_response()

impl CiClient {
    pub async fn protocol(
        &self,
        addr: &SocketAddr,
        config: Arc<thrussh::client::Config>,
        key: Arc<thrussh_keys::key::KeyPair>,
        receiver: &mut tokio::sync::mpsc::Receiver<(ci::Job, Option<i32>, Option<PathBuf>)>,
    ) -> Result<(), anyhow::Error> {
        let mut h = thrussh::client::connect(config, &addr, self.clone()).await?;
        debug!("Opening session");
        if !h.authenticate_publickey("ci", key).await? {
            return Ok(());
        }
        let mut channel = h.channel_open_session().await?;
        channel
            .data(
                &bincode::serialize(&Message::Handshake {
                    version: ci::VERSION,
                    id: 0,
                })
                .unwrap()[..],
            )
            .await?;
        debug!("handshake done");
        'outer: loop {
            if self.process.lock().unwrap().is_ready() {
                channel
                    .data(&bincode::serialize(&Message::Ready).unwrap()[..])
                    .await?;
                debug!("ready");
            }
            loop {
                tokio::select! {
                    msg = channel.wait() => {
                        debug!("msg = {:?}", msg);
                        if !self.handle_msg(&mut channel, &self.sender, msg).await? {
                            break 'outer
                        }
                    }
                    msg = receiver.recv() => {
                        debug!("message {:#?}", msg);
                        if let Some(p) = self.process.lock().unwrap().child.take() {
                            p.await??
                        }
                        if let Some((job, exit_status, path)) = msg {
                            self.send_log(&mut channel, job, exit_status, path).await?
                        }
                        channel.data(&bincode::serialize(&Message::Ready).unwrap()[..]).await?;
                    }
                }
            }
        }
        Ok(())

async fn stderr(
    State(config): State<Config>,
    Path(job): Path<uuid::Uuid>,
    range: Option<TypedHeader<Range>>,
) -> Response {
    debug!("stderr {:?} {:?}", job, range);
    let Ok(mut file) =
        tokio::fs::File::open(&config.config.ci_path.join(&format!("{}.stderr", job))).await
    else {
        return StatusCode::NOT_FOUND.into_response();
    };
    if let Some(TypedHeader(range)) = range {
        let body = KnownSize::file(file).await.unwrap();
        Ranged::new(Some(range), body).into_response()
    } else {
        use tokio::io::AsyncReadExt;
        let mut v = Vec::new();
        file.read_to_end(&mut v).await.unwrap();
        v.into_response()

}

    async fn handle_msg(
        &self,
        channel: &mut thrussh::client::Channel,
        sender: &tokio::sync::mpsc::Sender<(ci::Job, Option<i32>, Option<PathBuf>)>,
        msg: Option<thrussh::ChannelMsg>,
    ) -> Result<bool, anyhow::Error> {
        match msg {
            Some(thrussh::ChannelMsg::Data { data }) => {
                let mut proc = self.process.lock().unwrap();
                if let Some((mut f, mut len)) = proc.tarball.take() {
                    debug!("len = {:?}", len);
                    use std::io::Write;
                    f.write_all(&data)?;
                    len -= data.len();
                    if len > 0 {
                        proc.tarball = Some((f, len));
                    }
                    return Ok(true);
                }
                let msg = bincode::deserialize::<Message>(&data);
                debug!("msg = {:?}", msg);
                match msg {
                    Ok(Message::Job(job)) => {
                        self.handle_job(channel, sender.clone(), &mut proc, job)
                            .await?
                    }
                    Ok(Message::Chunk { id, len, .. }) => {
                        let p = self.tarball_path.join(&format!("{}.tar.gz.tmp", id));
                        if len == 0 {
                            let p2 = self.tarball_path.join(&format!("{}.tar.gz", id));
                            std::fs::rename(&p, &p2)?;
                            proc.tarball = None;
                            let job = proc.job.take().unwrap();
                            self.handle_job(channel, sender.clone(), &mut proc, job)
                                .await?;
                            return Ok(true);
                        }
                        let file = std::fs::OpenOptions::new()
                            .write(true)
                            .create(true)
                            .append(true)
                            .open(&p)
                            .unwrap();
                        proc.tarball = Some((file, len as usize));
                    }
                    Ok(msg) => {
                        debug!("msg = {:?}", msg);
                    }
                    _ => return Ok(false),
                }
            }
            None => return Ok(false),
            msg => debug!("{:?}", msg),

#[derive(Error, Debug)]
pub enum Error {
    #[error("Lock")]
    Lock,
    #[error("Forbidden")]
    Forbidden,
    #[error("Not found")]
    NotFound,
    #[error(transparent)]
    Output(
        #[from]
        pijul_core::output::OutputError<
            <C as pijul_core::changestore::ChangeStore>::Error,
            pijul_core::pristine::sanakirja::MutTxn0,
            std::io::Error,
        >,
    ),
    #[error(transparent)]
    Reqwest(#[from] reqwest::Error),
    #[error(transparent)]
    Sanakirja(#[from] pijul_core::pristine::sanakirja::SanakirjaError),
    #[error(transparent)]
    IO(#[from] std::io::Error),
    #[error(transparent)]
    Json(#[from] serde_json::Error),
}
type C = pijul_core::changestore::filesystem::FileSystem;
impl IntoResponse for Error {
    fn into_response(self) -> Response {
        debug!("response {:?}", self);
        match self {
            Error::Forbidden => (StatusCode::FORBIDDEN, "{}").into_response(),
            Error::NotFound => (StatusCode::NOT_FOUND, "{}").into_response(),
            _ => (StatusCode::INTERNAL_SERVER_ERROR, "{}").into_response(),

        Ok(true)

}

    async fn handle_job(
        &self,
        channel: &mut thrussh::client::Channel,
        sender: tokio::sync::mpsc::Sender<(ci::Job, Option<i32>, Option<PathBuf>)>,
        process: &mut Process,
        job: ci::Job,
    ) -> Result<(), anyhow::Error> {
        let p = self.tarball_path.join(&format!("{}.tar.gz", job.id));
        debug!("p = {:?}", p);
        if std::fs::metadata(&p).is_err() {
            debug!("getting tarball");
            channel
                .data(&bincode::serialize(&Message::GetTarball { id: job.id }).unwrap()[..])
                .await?;
            process.job = Some(job);
            return Ok(());
        }
        debug!("tar = {:?}", p);
        let status = std::process::Command::new("tar")
            .args(&["-xf", p.to_str().unwrap()])
            .current_dir(&self.tarball_path)
            .status()
            .unwrap();
        debug!("nix: {:?}", status);

// #[derive(Debug, Clone)]
// struct RepoPath {
//     path: PathBuf,
//     remove_dir: bool,
//     remove_dot: bool,
// }

        let tarballp = self.tarball_path.join(job.id.to_string());
        let logp = self.log_path.clone();

#[axum::debug_handler]
pub async fn trigger(
    State(config): State<Config>,
    Json(t): Json<Trigger>,
) -> Result<StatusCode, Error> {
    tokio::spawn(async move {
        info!("trigger: {:?}", t);
        let t0 = std::time::Instant::now();
        let mut remote = pijul_remote::RemoteRepo::Http(pijul_remote::http::Http {
            url: format!("{}/{}/{}", config.config.nest_url, t.owner, t.repo)
                .parse()
                .unwrap(),
            channel: "main".to_string(),
            client: reqwest::ClientBuilder::new().build()?,
            headers: vec![],
            name: t.state.clone(),
        });

        let result_path = logp.join(&format!("{}.result", job.id));
        if let Ok(mut f) = std::fs::File::open(&result_path) {
            if let Ok(build_result) = serde_json::from_reader::<_, BuildResult>(&mut f) {
                sender
                    .send((build_result.job, build_result.status, build_result.link))
                    .await?;
                return Ok(());
            }
        }

        let path = tempfile::tempdir().unwrap();
        let pijul_config = pijul_config::Config::load(None, Vec::new()).unwrap();
        let repo = tokio::sync::Mutex::new(
            pijul_repository::Repository::init(&pijul_config, Some(&path.path()), None, None)
                .unwrap(),
        );
        {
            let mut repo = repo.lock().await;
            let txn = repo.pristine.arc_txn_begin()?;
            let mut channel = txn.write().open_or_create_channel("main")?;

        debug!("p = {:?}", tarballp);
        process.child = Some(tokio::task::spawn(async move {
            let mut process = tokio::process::Command::new("nix-build")
                .arg("default.nix")
                .current_dir(&tarballp)
                .stdin(std::process::Stdio::null())
                .stdout(std::process::Stdio::piped())
                .stderr(std::process::Stdio::piped())
                .spawn()

            let h = t.state.parse().unwrap();
            remote
                .clone_state(&mut repo, &txn, &mut channel, h)
                .await

            let stdout = process.stdout.as_mut().unwrap();
            let stderr = process.stderr.as_mut().unwrap();
            let mut fstdout =
                tokio::fs::File::create(logp.join(&format!("{}.stdout", job.id))).await?;
            let mut fstderr =
                tokio::fs::File::create(logp.join(&format!("{}.stderr", job.id))).await?;
            let (a, b) = futures::future::join(
                tokio::io::copy(stdout, &mut fstdout),
                tokio::io::copy(stderr, &mut fstderr),
            )
            .await;
            a?;
            b?;
            let status = process.wait().await?;
            debug!("status = {:?}", status);

            let mut result_file = std::fs::File::create(&result_path)?;
            let link = std::fs::read_link(&tarballp.join("result")).ok();
            serde_json::to_writer(
                &mut result_file,
                &BuildResult {
                    finished: chrono::Utc::now(),
                    status: status.code(),
                    job: job.clone(),
                    link: link.clone(),
                },

            pijul_core::output::output_repository_no_pending(
                &repo.working_copy,
                &repo.changes,
                &txn,
                &channel,
                "",
                true,
                None,
                1, // std::thread::available_parallelism()?.get(),
                0,

            sender.send((job, status.code(), link)).await?;

            remote.finish().await.unwrap();
        }
        info!("cloned in {:?}", t0.elapsed());

            std::fs::remove_dir_all(&tarballp).unwrap_or(());
            std::fs::remove_file(&p)?;
            Ok(())
        }));
        Ok(())
    }

        std::fs::remove_dir_all(&path.path().join(".pijul"))?;

    async fn send_log(
        &self,
        channel: &mut thrussh::client::Channel,
        job: ci::Job,
        exit_status: Option<i32>,
        path: Option<PathBuf>,
    ) -> Result<(), anyhow::Error> {
        let id = job.id;
        let msg = Message::Log {
            job,
            exit_status,

        let (status_tx, status_rx) = tokio::sync::watch::channel(None);
        let (kill_tx, kill_rx) = tokio::sync::oneshot::channel();
        if let Some((k, _)) = config
            .jobs
            .lock()
            .unwrap()
            .insert(t.id, (kill_tx, status_rx))
        {
            k.send(()).unwrap_or(());
        }
        let result = build(
            Some(&config.config.ci_path),

        };
        channel.data(&bincode::serialize(&msg).unwrap()[..]).await?;

            t.id,
            &t.targets,
            kill_rx,
            status_tx,
        )
        .await;
        config.jobs.lock().unwrap().remove(&t.id);

        let mut buf = Vec::with_capacity(4096);
        debug!(
            "stdout: {:?}",
            self.log_path.join(&format!("{}.stdout", id))
        );
        if let Ok(ref mut stdout) =
            std::fs::File::open(&self.log_path.join(&format!("{}.stdout", id)))
        {
            let len = channel.writable_packet_size().min(MAX_BUF_SIZE);
            buf.resize(len, 0);
            while let Ok(n) = stdout.read(&mut buf) {
                if n == 0 {
                    channel
                        .data(
                            &bincode::serialize(&Message::Chunk {
                                id,
                                stderr: false,
                                len: 0,
                            })
                            .unwrap()[..],
                        )
                        .await?;
                    break;
                }
                channel
                    .data(
                        &bincode::serialize(&Message::Chunk {
                            id,
                            stderr: false,
                            len: n as u32,
                        })
                        .unwrap()[..],
                    )
                    .await?;
                channel.data(&buf[..n]).await?
            }
        }
        if let Ok(ref mut stdout) =
            std::fs::File::open(&self.log_path.join(&format!("{}.stderr", id)))
        {
            let len = channel.writable_packet_size().min(MAX_BUF_SIZE);
            buf.resize(len, 0);
            while let Ok(n) = stdout.read(&mut buf) {
                if n == 0 {
                    channel
                        .data(
                            &bincode::serialize(&Message::Chunk {
                                id,
                                stderr: true,
                                len: 0,
                            })
                            .unwrap()[..],
                        )
                        .await?;
                    break;
                }
                channel
                    .data(
                        &bincode::serialize(&Message::Chunk {
                            id,
                            stderr: true,
                            len: n as u32,
                        })
                        .unwrap()[..],
                    )
                    .await?;
                channel.data(&buf[..n]).await?
            }
        }
        Ok(())
    }

        let (code, results) = result?;
        debug!("code = {:?} {:?}", code, results);
        tokio::fs::write(
            config.config.ci_path.join(&format!("{}.status", t.id)),
            serde_json::to_string(&Status {
                code,
                results,
                finished: chrono::Utc::now(),
            })
            .unwrap(),
        )
        .await?;
        Ok::<(), Error>(())
    });
    Ok(StatusCode::OK)

const MAX_BUF_SIZE: usize = 1 << 16;

async fn build(
    ci: Option<&std::path::Path>,
    path: tempfile::TempDir,
    id: uuid::Uuid,
    targets: &HashMap<String, String>,
    mut kill_rx: tokio::sync::oneshot::Receiver<()>,
    mut status_tx: tokio::sync::watch::Sender<Option<i32>>,
) -> Result<(Option<i32>, HashMap<String, PathBuf>), Error> {
    let mut files = if let Some(ref ci_path) = ci {
        tokio::fs::create_dir_all(ci_path).await?;
        Some((
            OpenOptions::new()
                .append(true)
                .create(true)
                .open(&ci_path.join(&format!("{}.stdout", id)))
                .await
                .unwrap(),
            OpenOptions::new()
                .append(true)
                .create(true)
                .open(&ci_path.join(&format!("{}.stderr", id)))
                .await
                .unwrap(),
        ))
    } else {
        None
    };

impl thrussh::client::Handler for CiClient {
    type Error = anyhow::Error;

    let mut cmd = tokio::process::Command::new("nix");

    fn check_server_key(
        self,
        server_public_key: &thrussh_keys::key::PublicKey,
    ) -> impl futures::Future<Output = Result<(Self, bool), Self::Error>> {
        let valid = self
            .server_public_keys
            .iter()
            .any(|p| p == server_public_key);
        futures::future::ready(Ok((self, valid)))

    cmd.arg("build").arg("-L").arg("--log-format").arg("raw");
    for (_, target) in targets {
        cmd.arg(target);

    let cmd = cmd
        .current_dir(&path)
        .stderr(Stdio::piped())
        .stdout(Stdio::piped())
        .stdin(Stdio::null())
        .spawn()
        .unwrap();
    let code = dump_cmd(cmd, &mut files, &mut kill_rx, &mut status_tx).await?;

    fn adjust_window(&mut self, _channel: thrussh::ChannelId, target: u32) -> u32 {
        let elapsed = self.last_window_adjustment.elapsed().unwrap();
        self.last_window_adjustment = SystemTime::now();
        if target >= 10_000_000 {
            return target;
        }
        if elapsed < Duration::from_secs(2) {
            target * 2
        } else if elapsed > Duration::from_secs(8) {
            target / 2
        } else {
            target

    if code == Some(0) {
        let mut cmd = tokio::process::Command::new("attic");
        cmd.arg("push").arg("coturnix");
        let mut result = HashMap::new();
        for (n, (t, _)) in targets.iter().enumerate() {
            if n == 0 {
                result.insert(
                    t.clone(),
                    tokio::fs::read_link(path.path().join("result"))
                        .await
                        .unwrap(),
                );
                cmd.arg("result");
            } else {
                result.insert(
                    t.clone(),
                    tokio::fs::read_link(path.path().join(format!("result-{n}")))
                        .await
                        .unwrap(),
                );
                cmd.arg(format!("result-{n}"));
            }

        let cmd = cmd
            .current_dir(&path)
            .stderr(Stdio::piped())
            .stdout(Stdio::piped())
            .stdin(Stdio::null())
            .spawn()
            .unwrap();
        Ok((
            dump_cmd(cmd, &mut files, &mut kill_rx, &mut status_tx).await?,
            result,
        ))
    } else {
        Ok((code, HashMap::new()))

use serde_derive::*;

use serde::*;
use std::collections::HashMap;
use tokio::io::AsyncWriteExt;
use tracing::*;

pub const VERSION: u16 = 0;

#[derive(Debug, Serialize, Deserialize)]
pub struct Status {
    pub code: Option<i32>,
    pub finished: chrono::DateTime<chrono::Utc>,
    pub results: HashMap<String, std::path::PathBuf>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Job {

#[derive(Debug, Serialize, Deserialize)]
pub struct Trigger {

    pub repo: uuid::Uuid,
    pub channel: String,
    pub state: libpijul::pristine::Merkle,
    pub extra_patches: Vec<libpijul::pristine::Hash>,

    pub owner: String,
    pub repo: String,
    pub state: String,
    #[serde(default)]
    pub targets: HashMap<String, String>,

#[derive(Debug, Serialize, Deserialize)]
pub enum Message {
    Handshake {
        version: u16,
        id: i64,
    },
    Ready,
    Job(Job),
    GetTarball {
        id: uuid::Uuid,
    },
    InvalidInput {
        input: String,
    },
    Log {
        job: Job,
        exit_status: Option<i32>,
        path: Option<std::path::PathBuf>,
    },
    Chunk {
        id: uuid::Uuid,
        stderr: bool,
        len: u32,
    },

pub async fn dump_cmd(
    mut cmd: tokio::process::Child,
    files: &mut Option<(tokio::fs::File, tokio::fs::File)>,
    mut kill_rx: &mut tokio::sync::oneshot::Receiver<()>,
    status_tx: &mut tokio::sync::watch::Sender<Option<i32>>,
) -> Result<Option<i32>, std::io::Error> {
    let mut stdout_ok = true;
    let mut stderr_ok = true;
    let mut buf_stdout = String::new();
    let mut last_stdout = std::time::UNIX_EPOCH;
    let mut buf_stderr = String::new();
    let mut last_stderr = std::time::UNIX_EPOCH;
    let bound = std::time::Duration::from_secs(1);
    use tokio::io::AsyncBufReadExt;
    let stdout = tokio::io::BufReader::new(cmd.stdout.take().unwrap());
    let mut stdout = stdout.lines();
    let stderr = tokio::io::BufReader::new(cmd.stderr.take().unwrap());
    let mut stderr = stderr.lines();
    while stdout_ok || stderr_ok {
        debug!(
            "stdout || stderr {:?} {:?}",
            buf_stdout.len(),
            buf_stderr.len()
        );
        tokio::select! {
            line = stdout.next_line(), if stdout_ok => {
                debug!("out {:?}", line);
                let n = if let Some(line) = line? {
                    buf_stdout.push_str(&line);
                    buf_stdout.push('\n');
                    line.len() + 1
                } else {
                    0
                };
                if last_stdout.elapsed().unwrap() >= bound  || n == 0 {
                    debug!("sending stdout to db {:?} {:?}", buf_stdout.len(), buf_stderr.len());
                    if let Some((ref mut stdout, _)) = files {
                        stdout.write_all(buf_stdout.as_bytes()).await?;
                        stdout.flush().await?;
                    }
                    buf_stdout.clear();
                    debug!("stdout/stderr {:?} {:?}", buf_stdout.len(), buf_stderr.len());
                    last_stdout = std::time::SystemTime::now();
                }
                if n == 0 {
                    stdout_ok = false
                }
            }
            line = stderr.next_line(), if stderr_ok => {
                debug!("err {:?}", line);
                let n = if let Some(line) = line? {
                    buf_stderr.push_str(&line);
                    buf_stderr.push('\n');
                    line.len() + 1
                } else {
                    0
                };
                if last_stderr.elapsed().unwrap() >= bound || n == 0 {
                    debug!("sending stderr to db {:?}", buf_stderr.len());
                    if let Some((_, ref mut stderr)) = files {
                        stderr.write_all(buf_stderr.as_bytes()).await?;
                        stderr.flush().await?;
                    }
                    buf_stderr.clear();
                    last_stderr = std::time::SystemTime::now();
                }
                debug!("{:?}", buf_stderr.len());
                if n == 0 {
                    stderr_ok = false
                }
            }
            _ = &mut kill_rx => {
                cmd.kill().await?;
                break
            }
        }
    }
    tokio::select! {
        s = cmd.wait() => {
            let code = s?.code();
            status_tx.send(code).unwrap();
            Ok(code)
        }
        _ = &mut kill_rx => {
            cmd.kill().await?;
            Ok(None)
        }
    }

anyhow = "1.0.100"
bincode = "1.0"
chrono = "0.4.42"
clap = { version = "4.5.53", features = ["derive"] }
env_logger = "0.11.8"
futures = "0.3.31"
log = "0.4.28"
serde = "1.0.228"
serde_derive = "1.0.228"
serde_json = "1.0.145"
thrussh = "0.40"
thrussh-keys = "0.23"

anyhow.workspace = true
bincode.workspace = true
chrono.workspace = true
clap.workspace = true
env_logger.workspace = true
futures.workspace = true
log.workspace = true
serde.workspace = true
serde_derive.workspace = true
serde_json.workspace = true
thrussh.workspace = true
thrussh-keys.workspace = true

libpijul = { version = "1.0.0-beta.10", features = ["tarball"] }

pijul-core.workspace = true

tokio = { version = "1.48", features = [ "process", "fs" ] }
toml = { version = "0.9.8" }
uuid = { version = "1.21.0", features = ["serde" ] }
tracing = "0.1.41"
tracing-subscriber = { version = "0.3.20", features = ["env-filter"] }

tokio.workspace = true
toml.workspace = true
uuid.workspace = true
tracing.workspace = true
tracing-subscriber.workspace = true
axum.workspace = true
thiserror.workspace = true
pijul-remote.workspace = true
reqwest.workspace = true
pijul-repository.workspace = true
pijul-config.workspace = true
sanakirja.workspace = true
pijul-interaction.workspace = true
tempfile.workspace = true
diesel.workspace = true
diesel-async.workspace = true
axum-range.workspace = true
axum-extra = { workspace = true, features = ["typed-header"] }

use libpijul::TxnTExt;
use libpijul::pristine::{Base32, ChannelTxnT, DepsTxnT, GraphTxnT, TxnT};

use pijul_core::TxnTExt;
use pijul_core::pristine::{Base32, ChannelTxnT, DepsTxnT, GraphTxnT, TxnT};

    pub applied: Vec<libpijul::pristine::Hash>,

    pub applied: Vec<pijul_core::pristine::Hash>,

    key: Option<libpijul::key::PublicKey>,

    key: Option<pijul_core::key::PublicKey>,

        hash: libpijul::pristine::Hash,

        hash: pijul_core::pristine::Hash,

impl From<libpijul::key::Algorithm> for Keyalgorithm_ {
    fn from(a: libpijul::key::Algorithm) -> Keyalgorithm_ {

impl From<pijul_core::key::Algorithm> for Keyalgorithm_ {
    fn from(a: pijul_core::key::Algorithm) -> Keyalgorithm_ {

            libpijul::key::Algorithm::Ed25519 => Keyalgorithm_::Ed25519,

            pijul_core::key::Algorithm::Ed25519 => Keyalgorithm_::Ed25519,

impl From<Keyalgorithm_> for libpijul::key::Algorithm {
    fn from(a: Keyalgorithm_) -> libpijul::key::Algorithm {

impl From<Keyalgorithm_> for pijul_core::key::Algorithm {
    fn from(a: Keyalgorithm_) -> pijul_core::key::Algorithm {

            Keyalgorithm_::Ed25519 => libpijul::key::Algorithm::Ed25519,

            Keyalgorithm_::Ed25519 => pijul_core::key::Algorithm::Ed25519,

        hash: libpijul::pristine::Hash,

        hash: pijul_core::pristine::Hash,

        if let Ok(json) = serde_json::from_slice::<libpijul::key::PublicKey>(cap[1].as_bytes()) {

        if let Ok(json) = serde_json::from_slice::<pijul_core::key::PublicKey>(cap[1].as_bytes()) {

                    ChannelSpec::CI(ref c_) => {
                        if c_.len() + 3 == cap[1].len() {
                            let (a, b) = cap[1].split_at(cap[1].len() - 3);
                            if c_ == a && b == "/ci" {
                                return Ok(ChannelList::Data(vec![b'\n']));
                            }
                        }
                    }

                    let h: libpijul::Hash = h.into();

                    let h: pijul_core::Hash = h.into();

                        libpijul::fs::iter_graph_descendants(&txn, txn.graph(&*channel.read()), p)?

                        pijul_core::fs::iter_graph_descendants(&txn, txn.graph(&*channel.read()), p)?

                    let h: libpijul::Hash = h.into();
                    let m: libpijul::Merkle = m.into();

                    let h: pijul_core::Hash = h.into();
                    let m: pijul_core::Merkle = m.into();

                        let m: libpijul::Merkle = m.into();

                        let m: pijul_core::Merkle = m.into();

                            libpijul::Merkle::zero()

                            pijul_core::Merkle::zero()

                    let m: libpijul::Merkle = m.into();

                    let m: pijul_core::Merkle = m.into();

                        libpijul::Merkle::zero()

                        pijul_core::Merkle::zero()

        let hash = if let Some(h) = libpijul::pristine::Hash::from_base32((&cap[2]).as_bytes()) {

        let hash = if let Some(h) = pijul_core::pristine::Hash::from_base32((&cap[2]).as_bytes()) {

        let hash = if let Some(h) = libpijul::pristine::Hash::from_base32((&cap[4]).as_bytes()) {

        let hash = if let Some(h) = pijul_core::pristine::Hash::from_base32((&cap[4]).as_bytes()) {

        libpijul::changestore::filesystem::push_filename(&mut p, &hash);

        pijul_core::changestore::filesystem::push_filename(&mut p, &hash);

            libpijul::change::Change::size_no_contents(&mut f)?

            pijul_core::change::Change::size_no_contents(&mut f)?

        hash: libpijul::pristine::Hash,

        hash: pijul_core::pristine::Hash,

            Some(ChannelSpec::CI(_)) | Some(ChannelSpec::Channel(_)) => {

            Some(ChannelSpec::Channel(_)) => {

            ChannelSpec::CI(c) => self.channel = Some(ChannelSpec::CI(c)),

        hash: libpijul::Hash,

        hash: pijul_core::Hash,

        use libpijul::changestore::ChangeStore;

        use pijul_core::changestore::ChangeStore;

        hash: libpijul::Hash,

        hash: pijul_core::Hash,

                        use libpijul::changestore::ChangeStore;

                        use pijul_core::changestore::ChangeStore;

            use libpijul::changestore::ChangeStore;

            use pijul_core::changestore::ChangeStore;

        hash: libpijul::pristine::Hash,
        header: libpijul::change::ChangeHeader,

        hash: pijul_core::pristine::Hash,
        header: pijul_core::change::ChangeHeader,

                    r::creation_ip.eq(addr.to_string()),

                    // r::creation_ip.eq(addr.ip()),

use libpijul::{

use pijul_core::{

    channel: Option<String>,

    meta: libpijul::pristine::InodeMetadata,

    meta: pijul_core::pristine::InodeMetadata,

    pos: Query<PosQuery>,

    Query(pos): Query<PosQuery>,

    let (id, _) = super::repository_id(&mut db, &tree.owner, &tree.repo, uid, Perm::READ).await?;

    let (id, _, _) = super::repository_id(&mut db, &tree.owner, &tree.repo, uid, Perm::READ).await?;

        } else if txn.channels("")?.is_empty()
            && (pos.channel.as_deref() == Some("main") || pos.channel.is_none())
        {

        } else if txn.channels("")?.is_empty() && channel_ == "main" {

            for x in libpijul::fs::iter_graph_children(&txn, &repo_.changes, txn.graph(&*ch), pos)?

            for x in
                pijul_core::fs::iter_graph_children(&txn, &repo_.changes, txn.graph(&*ch), pos)?

            let txn = libpijul::ArcTxn::new(txn);
            libpijul::output::output_file(&repo_.changes, &txn, &channel, pos, &mut out)

            let txn = pijul_core::ArcTxn::new(txn);
            pijul_core::output::output_file(&repo_.changes, &txn, &channel, pos, &mut out)

        _: Option<(&T, &[&libpijul::Hash])>,

        _: Option<(&T, &[&pijul_core::Hash])>,

use libpijul::{Base32, MutTxnT};

use pijul_core::{Base32, MutTxnT};

pub mod dot_pijul;

    pub pristine: RwLock<libpijul::pristine::sanakirja::Pristine>,

    pub pristine: RwLock<pijul_core::pristine::sanakirja::Pristine>,

    ) -> Result<libpijul::pristine::sanakirja::Pristine, crate::Error> {

    ) -> Result<pijul_core::pristine::sanakirja::Pristine, crate::Error> {

            let repo = libpijul::pristine::sanakirja::Pristine::new_nolock(&path.join("db"))?;

            let repo = pijul_core::pristine::sanakirja::Pristine::new_nolock(&path.join("db"))?;

            Ok(libpijul::pristine::sanakirja::Pristine::new_nolock(

            Ok(pijul_core::pristine::sanakirja::Pristine::new_nolock(

#[derive(Debug)]

#[derive(Debug, PartialEq, Eq)]

    CI(String),

        if c.ends_with("/ci") {
            ChannelSpec::CI(c.split_at(c.len() - 3).0.to_string())
        } else {

        }

) -> Result<(Uuid, Perm), crate::Error> {

) -> Result<(Uuid, Uuid, Perm), crate::Error> {

    if let Some((rid, perms, suspended, is_public)) = r::repositories

    if let Some((rid, perms, uid, suspended, is_public)) = r::repositories

            u::id,

        .get_result::<(uuid::Uuid, i64, bool, bool)>(db)

        .get_result::<(uuid::Uuid, i64, uuid::Uuid, bool, bool)>(db)

            Ok((rid, got))

            Ok((rid, uid, got))

        use libpijul::TxnT;

        use pijul_core::TxnT;

    pub public_key: libpijul::key::PublicKey,

    pub public_key: pijul_core::key::PublicKey,

    pub meta: libpijul::pristine::InodeMetadata,

    pub meta: pijul_core::pristine::InodeMetadata,

    T: libpijul::GraphTxnT + libpijul::ChannelTxnT,
    C: libpijul::changestore::ChangeStore,

    T: pijul_core::GraphTxnT + pijul_core::ChannelTxnT,
    C: pijul_core::changestore::ChangeStore,

    pos: libpijul::pristine::Position<libpijul::ChangeId>,

    pos: pijul_core::pristine::Position<pijul_core::ChangeId>,

            if let Some(x) = libpijul::fs::iter_basenames(txn, changes, txn.graph(channel), pos)

            if let Some(x) = pijul_core::fs::iter_basenames(txn, changes, txn.graph(channel), pos)

use libpijul::change::{Change, ChangeFile, ChangeHeader};
use libpijul::changestore::filesystem::*;
use libpijul::changestore::*;
use libpijul::pristine::{Base32, ChangeId, Hash, Merkle, Position, Vertex};

use pijul_core::change::{Change, ChangeFile, ChangeHeader};
use pijul_core::changestore::filesystem::*;
use pijul_core::changestore::*;
use pijul_core::pristine::{Base32, ChangeId, Hash, Merkle, Position, Vertex};

    Tag(#[from] libpijul::tag::TagError),

    Tag(#[from] pijul_core::tag::TagError),

    ChangeFile(#[from] libpijul::change::ChangeError),

    ChangeFile(#[from] pijul_core::change::ChangeError),

        libpijul::change::ChangeError,

        pijul_core::change::ChangeError,

                let p = Arc::new(Mutex::new(libpijul::change::ChangeFile::open(

                let p = Arc::new(Mutex::new(pijul_core::change::ChangeFile::open(

    ) -> Result<(), libpijul::change::ChangeError> {

    ) -> Result<(), pijul_core::change::ChangeError> {

use libpijul::change::*;

use pijul_core::change::*;

) -> Result<Hashed<Hunk<Option<Hash>, Local>, Author>, libpijul::change::ChangeError> {
    use libpijul::change::*;
    use libpijul::pristine::Hasher;

) -> Result<Hashed<Hunk<Option<Hash>, Local>, Author>, pijul_core::change::ChangeError> {
    use pijul_core::change::*;
    use pijul_core::pristine::Hasher;

        if let Ok(p) = libpijul::change::ChangeFile::open(hash, &path.to_str().unwrap()) {

        if let Ok(p) = pijul_core::change::ChangeFile::open(hash, &path.to_str().unwrap()) {

        let p = libpijul::change::ChangeFile::open(*h, &path.to_str().unwrap())?;

        let p = pijul_core::change::ChangeFile::open(*h, &path.to_str().unwrap())?;

        let mut p = libpijul::tag::OpenTagFile::open(&path, h)?;

        let mut p = pijul_core::tag::OpenTagFile::open(&path, h)?;

            let mut p = libpijul::change::ChangeFile::open(change, &path.to_str().unwrap())?;

            let mut p = pijul_core::change::ChangeFile::open(change, &path.to_str().unwrap())?;

        E: From<Self::Error> + From<libpijul::change::ChangeError>,

        E: From<Self::Error> + From<pijul_core::change::ChangeError>,

    let (id, _) = crate::repository::repository_id(

    let (id, _, _) = crate::repository::repository_id(

    let (id, _) = crate::repository::repository_id(

    let (id, _, _) = crate::repository::repository_id(

use libpijul::changestore::ChangeStore;
use libpijul::fs::FsErrorC;
use libpijul::output::{FileError, OutputError};
use libpijul::pristine::sanakirja::MutTxn0;
use libpijul::pristine::sanakirja::SanakirjaError;
use libpijul::pristine::{ForkError, TreeErr, TxnErr};
use libpijul::{
    ApplyError, ArcTxn, ChannelMutTxnT, ChannelRef, MutTxnT, MutTxnTExt, TxnT, TxnTExt,
    UnrecordError,

use pijul_core::changestore::ChangeStore;
use pijul_core::fs::FsErrorC;
use pijul_core::output::{FileError, OutputError};
use pijul_core::pristine::sanakirja::MutTxn0;
use pijul_core::pristine::sanakirja::SanakirjaError;
use pijul_core::pristine::{ForkError, TreeErr, TxnErr};
use pijul_core::{
    ApplyError, ArcTxn, Base32, ChannelMutTxnT, MutTxnT, MutTxnTExt, TxnT, TxnTExt, UnrecordError,

use std::process::Stdio;

    client: reqwest::Client,

            client: reqwest::Client::new(),

    #[error(transparent)]
    Reqwest(#[from] reqwest::Error),

                            Err(libpijul::ApplyError::LocalChange(
                                libpijul::LocalApplyError::ChangeAlreadyOnChannel { .. },

                            Err(pijul_core::ApplyError::LocalChange(
                                pijul_core::LocalApplyError::ChangeAlreadyOnChannel { .. },

                            Err(libpijul::ApplyError::LocalChange(
                                libpijul::LocalApplyError::DependencyMissing { hash },

                            Err(pijul_core::ApplyError::LocalChange(
                                pijul_core::LocalApplyError::DependencyMissing { hash },

                                    libpijul::ApplyError::LocalChange(
                                        libpijul::LocalApplyError::DependencyMissing { hash },

                                    pijul_core::ApplyError::LocalChange(
                                        pijul_core::LocalApplyError::DependencyMissing { hash },

                        let repo_ = s.locks.get(&repo).await.unwrap();
                        tokio::task::spawn_blocking(move || {
                            if let Some((temp, depl)) = {
                                let pri = repo_.pristine.blocking_write();
                                let txn = pri.arc_txn_begin()?;
                                let channel = format!("{}_{}", repo, channel);
                                let channel_ = {
                                    let mut txn_ = txn.write();
                                    txn_.open_or_create_channel(&channel)?
                                };
                                let changes = repo_.changes.clone();

                        if let Some((state, deployment)) =
                            get_config_state(&s.locks, repo, channel).await?
                        {
                            debug!("{:?} {:?}", state, deployment);
                            let mut db = s.db.get().await.unwrap();
                            use crate::db::jobs::dsl as jobs;
                            let id = diesel::insert_into(jobs::jobs)
                                .values((
                                    jobs::repo.eq(repo),
                                    jobs::repo_state.eq(state.to_base32()),
                                ))
                                .returning(jobs::id)
                                .get_result::<uuid::Uuid>(&mut db)
                                .await
                                .unwrap();
                            use crate::db::repositories::dsl as repositories;
                            use crate::db::users::dsl as users;

                                s.output_for_deployment(&txn, &channel_, &changes)?
                            } {
                                tokio::spawn(async move {
                                    use crate::db::jobs::dsl as jobs;
                                    let id = diesel::insert_into(jobs::jobs)
                                        .values((jobs::repo.eq(repo),))
                                        .returning(jobs::id)
                                        .get_result::<uuid::Uuid>(&mut s.db.get().await.unwrap())
                                        .await
                                        .unwrap();

                            let (owner, repo) = repositories::repositories
                                .find(repo)
                                .inner_join(users::users)
                                .select((users::login, repositories::name))
                                .get_result::<(String, String)>(&mut db)
                                .await
                                .unwrap();
                            let mut targets = std::collections::HashMap::new();
                            targets.insert(deployment.clone(), format!(".#{}", deployment));
                            let body = serde_json::to_string(&ci::Trigger {
                                id,
                                owner,
                                repo,
                                state: state.to_base32(),
                                targets,
                            })
                            .unwrap();
                            debug!("{:?} {}", s.ci.url, body);
                            for (n, ci) in s.ci.url.iter().enumerate() {
                                let res = s
                                    .client
                                    .post(ci.clone() + "/trigger")
                                    .header("Content-Type", "application/json")
                                    .body(body.clone())
                                    .send()
                                    .await
                                    .unwrap();
                                debug!("{:?}", res);

                                    let permit = s.builders.acquire().await.unwrap();
                                    {
                                        s.deploy(id, temp, depl).await?;
                                    }
                                    std::mem::drop(permit);
                                    Ok::<_, Error>(())
                                });

                                if let reqwest::StatusCode::OK = res.status() {
                                    tokio::spawn(async move {
                                        sync_job(&mut db, s.client, &s.ci, n, id).await;
                                    });
                                    break;
                                }

                            Ok::<_, Error>(())
                        });

                        }

                            &libpijul::working_copy::sink(),

                            &pijul_core::working_copy::sink(),

                            Err(libpijul::UnrecordError::ChangeNotInChannel { .. }) | Ok(_) => {

                            Err(pijul_core::UnrecordError::ChangeNotInChannel { .. }) | Ok(_) => {

                            libpijul::changestore::filesystem::push_filename(&mut p, &hash);

                            pijul_core::changestore::filesystem::push_filename(&mut p, &hash);

    txn: &ArcTxn<libpijul::pristine::sanakirja::MutTxn0>,
    channel: &ChannelRef<libpijul::pristine::sanakirja::MutTxn0>,

    txn: &ArcTxn<pijul_core::pristine::sanakirja::MutTxn0>,
    channel: &pijul_core::ChannelRef<pijul_core::pristine::sanakirja::MutTxn0>,

    let txn_ = txn.read();
    let channel_ = channel.read();
    let (pos, is_dir) = txn_.follow_oldest_path(changes, &channel, path)?;

    let (pos, is_dir) = txn.read().follow_oldest_path(changes, channel, path)?;

    use libpijul::ChannelTxnT;
    let mut graph = libpijul::alive::retrieve(&*txn_, txn_.graph(&*channel_), pos, false)?;

    use pijul_core::ChannelTxnT;
    let mut graph = {
        let txn = txn.read();
        pijul_core::alive::retrieve(&*txn, txn.graph(&*channel.read()), pos, false)?
    };

    std::mem::drop(channel_);
    std::mem::drop(txn_);
    libpijul::alive::output_graph(changes, &txn, &channel, &mut out, &mut graph, &mut forward)

    pijul_core::alive::output_graph(changes, &txn, &channel, &mut out, &mut graph, &mut forward)

impl H {
    fn output_for_deployment<
        C: ChangeStore<Error = crate::repository::changestore::Error> + Clone + Send + Sync + 'static,
    >(
        &self,
        txn: &ArcTxn<libpijul::pristine::sanakirja::MutTxn0>,
        channel: &ChannelRef<libpijul::pristine::sanakirja::MutTxn0>,
        changes: &C,
    ) -> Result<Option<(tempfile::TempDir, String)>, Error> {
        if let Some(config) = get_file(txn, channel, changes, "pijul.toml")? {
            debug!("config = {:?}", config);

pub async fn get_config_state(
    locks: &RepositoryLocks,
    repo: uuid::Uuid,
    channel: String,
) -> Result<Option<(pijul_core::Merkle, String)>, Error> {
    debug!("get_config_state");
    let repo_ = locks.get(&repo).await.unwrap();
    tokio::task::spawn_blocking(move || {
        let pri = repo_.pristine.blocking_write();
        let txn_ = pri.arc_txn_begin()?;
        let channel = format!("{}_{}", repo, channel);
        if let Some(channel) = txn_.read().load_channel(&channel)? {
            if let Some(config) = get_file(&txn_, &channel, &repo_.changes, "pijul.toml")? {
                debug!("config = {:?}", config);

            if let Ok(parsed) = toml::from_str::<Config>(&config) {
                debug!("{:?}", parsed);
                if let Some(depl) = parsed.deployment {
                    let txn = txn.clone();
                    let channel = channel.clone();
                    let changes = changes.clone();
                    let tmp_dir = tempfile::tempdir().unwrap();
                    let wc =
                        libpijul::working_copy::filesystem::FileSystem::from_root(tmp_dir.path());
                    libpijul::output::output_repository_no_pending(
                        &wc, &changes, &txn, &channel, "", true, None, 1, 0,
                    )
                    .unwrap();
                    return Ok(Some((tmp_dir, depl)));

                if let Ok(parsed) = toml::from_str::<Config>(&config) {
                    if let Some(dep) = parsed.deployment {
                        if let Some(Ok((_, (_, state)))) =
                            txn_.read().reverse_log(&channel.read(), None)?.next()
                        {
                            return Ok(Some((state.into(), dep)));
                        }
                    }

            }

            };

    }
    async fn deploy(
        &self,
        id: uuid::Uuid,
        tmp_dir: tempfile::TempDir,
        depl: String,
    ) -> Result<(), Error> {
        use std::process::Stdio;
        let (status_tx, status_rx) = tokio::sync::watch::channel(None);
        let (kill_tx, kill_rx) = tokio::sync::oneshot::channel();
        let p = tmp_dir.path().join(depl);
        debug!("launching {:?}", p);

    })
    .await
    .unwrap()
}

        let mut cmd = tokio::process::Command::new(p)
            .current_dir(tmp_dir.path())
            .stderr(Stdio::piped())
            .stdout(Stdio::piped())
            .stdin(Stdio::null())
            .spawn()

pub async fn sync_job(
    db: &mut diesel_async::AsyncPgConnection,
    client: reqwest::Client,
    ci: &crate::config_file::CiConfig,
    ci_n: usize,
    job: uuid::Uuid,
) {
    let mut files = if let Some(ref f) = ci.filesystem {
        let stdout = OpenOptions::new()
            .append(true)
            .create(true)
            .open(&f.join(&format!("{}.stdout", job)))
            .await
            .unwrap();
        let stderr = OpenOptions::new()
            .append(true)
            .create(true)
            .open(&f.join(&format!("{}.stderr", job)))
            .await

        Some((stdout, stderr))
    } else {
        None
    };

        use tokio::io::AsyncBufReadExt;
        let stdout = tokio::io::BufReader::new(cmd.stdout.take().unwrap());
        let mut stdout = stdout.lines();
        let mut stdout_ok = true;
        let stderr = tokio::io::BufReader::new(cmd.stderr.take().unwrap());
        let mut stderr = stderr.lines();
        self.jobs
            .lock()
            .unwrap()
            .insert(id, (kill_tx, status_tx.clone(), status_rx));
        let mut stderr_ok = true;
        let mut buf_stdout = String::new();
        let mut last_stdout = std::time::UNIX_EPOCH;
        let mut buf_stderr = String::new();
        let mut last_stderr = std::time::UNIX_EPOCH;
        let bound = std::time::Duration::from_secs(1);
        let mut files = if let Some(ref path) = self.ci.filesystem {
            Some((
                OpenOptions::new()
                    .append(true)
                    .create(true)
                    .open(&path.join(&format!("{}.stdout", id)))
                    .await
                    .unwrap(),
                OpenOptions::new()
                    .append(true)
                    .create(true)
                    .open(&path.join(&format!("{}.stderr", id)))
                    .await
                    .unwrap(),
            ))

    loop {
        debug!("Querying status");
        let status = client
            .get(format!("{}/status/{}", ci.url[ci_n], job))
            .send()
            .await
            .unwrap();
        let status = if let reqwest::StatusCode::OK = status.status() {
            let status: ci::Status = status.json().await.unwrap();
            debug!("STATUS {:?}", status);
            use crate::db::jobs::dsl as jobs;
            diesel::update(jobs::jobs.find(job))
                .set((
                    jobs::status.eq(status.code),
                    jobs::ended.eq(status.finished),
                ))
                .execute(db)
                .await
                .unwrap();
            Some(status)

        while stdout_ok || stderr_ok {
            debug!(
                "stdout || stderr {:?} {:?}",
                buf_stdout.len(),
                buf_stderr.len()

        if let Some((ref mut stdout, ref mut stderr)) = files {
            let stdout_len = stdout.metadata().await.unwrap().len();
            let stderr_len = stderr.metadata().await.unwrap().len();
            let (stdout_, stderr_) = tokio::join!(
                client
                    .get(format!("{}/stdout/{}", ci.url[ci_n], job))
                    .header("Range", format!("bytes={stdout_len}-*"))
                    .send(),
                client
                    .get(format!("{}/stderr/{}", ci.url[ci_n], job))
                    .header("Range", format!("bytes={stderr_len}-*"))
                    .send()

            tokio::select! {
                line = stdout.next_line(), if stdout_ok => {
                    let n = if let Some(line) = line? {
                        buf_stdout.push_str(&line);
                        buf_stdout.push('\n');
                        line.len() + 1
                    } else {
                        0
                    };
                    if last_stdout.elapsed().unwrap() >= bound  || n == 0 {
                        debug!("sending stdout to db {:?} {:?}", buf_stdout.len(), buf_stderr.len());
                        if let Some((ref mut stdout, _)) = files {
                            stdout.write_all(buf_stdout.as_bytes()).await?;
                            stdout.flush().await?;
                        }
                        buf_stdout.clear();
                        debug!("stdout/stderr {:?} {:?}", buf_stdout.len(), buf_stderr.len());
                        last_stdout = std::time::SystemTime::now();
                    }
                    if n == 0 {
                        stdout_ok = false
                    }
                }
                line = stderr.next_line(), if stderr_ok => {
                    let n = if let Some(line) = line? {
                        buf_stderr.push_str(&line);
                        buf_stderr.push('\n');
                        line.len() + 1
                    } else {
                        0
                    };
                    if last_stderr.elapsed().unwrap() >= bound || n == 0 {
                        debug!("sending stderr to db {:?}", buf_stderr.len());
                        if let Some((_, ref mut stderr)) = files {
                            stderr.write_all(buf_stderr.as_bytes()).await?;
                            stderr.flush().await?;
                        }
                        buf_stderr.clear();
                        last_stderr = std::time::SystemTime::now();
                    }
                    debug!("{:?}", buf_stderr.len());
                    if n == 0 {
                        stderr_ok = false
                    }
                }
            }

            let stdout_ = stdout_.unwrap();
            let stderr_ = stderr_.unwrap();
            debug!("{:?} {:?}", stdout_, stderr_);
            let bytes = stdout_.bytes().await.unwrap();
            stdout.write(&bytes).await.unwrap();
            let bytes = stderr_.bytes().await.unwrap();
            stderr.write(&bytes).await.unwrap();

        let status = tokio::select! {
            status = cmd.wait() => {
                status?.code()

        if let Some(status) = status {
            let (mut status_tx, _status_rx) = tokio::sync::watch::channel(None);
            let (_kill_tx, mut kill_rx) = tokio::sync::oneshot::channel();
            for (script, s) in status.results.iter() {
                let mut cmd = tokio::process::Command::new("nix-store");
                cmd.arg("-r").arg(&s);
                let cmd = cmd
                    .stderr(Stdio::piped())
                    .stdout(Stdio::piped())
                    .stdin(Stdio::null())
                    .spawn()
                    .unwrap();
                ci::dump_cmd(cmd, &mut files, &mut kill_rx, &mut status_tx)
                    .await
                    .unwrap();

            _ = kill_rx => {
                cmd.kill().await?;
                None

            for (script, s) in status.results.iter() {
                debug!("launching {:?} {:?}", s, script);
                let cmd = tokio::process::Command::new(format!("{}/bin/{script}", s.display()))
                    .stderr(Stdio::piped())
                    .stdout(Stdio::piped())
                    .stdin(Stdio::null())
                    .spawn()
                    .unwrap();
                ci::dump_cmd(cmd, &mut files, &mut kill_rx, &mut status_tx)
                    .await
                    .unwrap();

        };
        debug!("process exited with {:?}", status);
        debug!("stderr {}", buf_stderr);
        debug!("stdout {}", buf_stdout);
        let now = chrono::Utc::now();
        use crate::db::jobs::dsl as jobs;
        diesel::update(jobs::jobs.find(id))
            .set((jobs::status.eq(status), jobs::ended.eq(&now)))
            .execute(&mut self.db.get().await.unwrap())
            .await?;
        status_tx.send(Some((now, status))).unwrap();
        self.jobs.lock().unwrap().remove(&id);
        Ok::<_, Error>(())

            break;
        }
        tokio::time::sleep(std::time::Duration::from_secs(1)).await

    http::uri::Uri,

mod proxy;

// mod proxy;

        .route("/static/{*wildcard}", any(proxy::node_proxy_cached))
        .route("/_app/immutable/{*wildcard}", any(proxy::node_proxy_cached))

        // .route("/static/{*wildcard}", any(proxy::node_proxy_cached))
        // .route("/_app/immutable/{*wildcard}", any(proxy::node_proxy_cached))

        .route("/", any(root))

        // .route("/", any(root))
        .route(
            "/{owner}/{repo}/.pijul",
            any(repository::dot_pijul::dot_pijul),
        )

        .fallback(proxy::node_proxy_not_cached)

        // .fallback(proxy::node_proxy_not_cached)

        let listener = tokio::net::TcpListener::bind(http_addr).await.unwrap();

                redirect_http_to_https(
                    config_file.http.http_port,
                    config_file.http.https_port,
                    app_,

                axum::serve(
                    listener,
                    Router::new()
                        .route("/", get(redirect_http_to_https))
                        .into_make_service_with_connect_info::<SocketAddr>(),

                .await;
                Ok::<_, Error>(())

                .await

                .await?;
                Ok::<_, Error>(())

                .await

/*

 */

pub async fn redirect_http_to_https(http: u16, https: u16, mut app: Router<()>) {
    fn make_https(
        host: String,
        uri: http::Uri,
        http: u16,
        https: u16,
    ) -> Result<http::Uri, axum::BoxError> {
        let mut parts = uri.into_parts();
        parts.scheme = Some(axum::http::uri::Scheme::HTTPS);
        if parts.path_and_query.is_none() {
            parts.path_and_query = Some("/".parse().unwrap());
        }
        let https_host = host.replace(&http.to_string(), &https.to_string());
        parts.authority = Some(https_host.parse()?);
        Ok(http::Uri::from_parts(parts)?)
    }
    use axum::extract::ConnectInfo;
    use axum::response::IntoResponse;
    use axum_extra::extract::Host;
    let redirect = move |Host(host): Host,
                         uri: http::Uri,
                         ConnectInfo(addr): ConnectInfo<SocketAddr>,
                         r: http::Request<axum::body::Body>| async move {
        debug!("redirect {:?}", addr);
        if addr.ip().is_loopback() {
            debug!("is loopback");
            use tower_service::Service;
            Ok(app.call(r).await.into_response())
        } else {
            debug!("not loopback");
            match make_https(host, uri, http, https) {
                Ok(uri) => Ok(Redirect::permanent(&uri.to_string()).into_response()),
                Err(error) => {
                    tracing::warn!(%error, "failed to convert URI to HTTPS");
                    Err(StatusCode::BAD_REQUEST.into_response())
                }
            }
        }
    };
    let addr = Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 0);
    let http_addr = SocketAddr::V6(SocketAddrV6::new(addr, http, 0, 0));
    let listener = tokio::net::TcpListener::bind(http_addr).await.unwrap();
    tracing::debug!("listening on {}", listener.local_addr().unwrap());
    use axum::handler::HandlerWithoutStateExt;
    axum::serve(
        listener,
        redirect.into_make_service_with_connect_info::<SocketAddr>(),
    )
    .await
    .unwrap();

async fn redirect_http_to_https(uri: Uri) -> Redirect {
    let uri = format!("https://{}{}", uri.host().unwrap(), uri.path());
    Redirect::temporary(&uri)

                let perm = Perm::from_bits(perm).unwrap();
                let perm_all = Perm::from_bits(perm_all).unwrap();

                let perm = Perm::from_bits_truncate(perm);
                let perm_all = Perm::from_bits_truncate(perm_all);

        libpijul::output::ArchiveError<

        pijul_core::output::ArchiveError<

            libpijul::pristine::sanakirja::GenericTxn<sanakirja::MutTxn<Arc<sanakirja::Env>>>,

            pijul_core::pristine::sanakirja::GenericTxn<sanakirja::MutTxn<Arc<sanakirja::Env>>>,

        libpijul::ApplyError<

        pijul_core::ApplyError<

            libpijul::pristine::sanakirja::GenericTxn<sanakirja::MutTxn<Arc<sanakirja::Env>>>,

            pijul_core::pristine::sanakirja::GenericTxn<sanakirja::MutTxn<Arc<sanakirja::Env>>>,

    PijulChange(#[from] libpijul::change::ChangeError),

    PijulChange(#[from] pijul_core::change::ChangeError),

    Sanakirja(#[from] libpijul::pristine::sanakirja::SanakirjaError),

    Sanakirja(#[from] pijul_core::pristine::sanakirja::SanakirjaError),

    Key(#[from] libpijul::key::KeyError),

    Key(#[from] pijul_core::key::KeyError),

impl<E: Into<Error> + std::error::Error> From<libpijul::pristine::TxnErr<E>> for Error {
    fn from(e: libpijul::pristine::TxnErr<E>) -> Self {

impl<E: Into<Error> + std::error::Error> From<pijul_core::pristine::TxnErr<E>> for Error {
    fn from(e: pijul_core::pristine::TxnErr<E>) -> Self {

use axum::response::IntoResponse;

    response::Response,
    routing::{any, get},

    response::{Redirect, Response},
    routing::{any, get, post},

use pijul_core::Base32;

use std::collections::hash_map::Entry;

        .route("/{owner}/{repo}/{job_id}/ws", any(ws_handler))

        .route("/{owner}/{repo}/{job_id}/retry", post(retry))
        .route("/{owner}/{repo}/{job_id}/kill", post(kill))
        .route("/{owner}/{repo}/{job_id}/ws", any(job_ws_handler))
        .route("/{owner}/{repo}/ws", any(jobs_ws_handler))

#[derive(Debug, Selectable, Queryable, QueryableByName, Serialize)]

#[derive(Debug, Selectable, Queryable, QueryableByName, Serialize, Deserialize)]

pub async fn retry(
    State(config): State<Config>,
    jar: SignedCookieJar,
    Path(path): Path<JobPath>,
) -> Result<Redirect, crate::Error> {
    let (uid, login) = if let Some((a, b)) = get_user_login(&jar, &config).await? {
        (Some(a), Some(b))
    } else {
        (None, None)
    };
    let mut db = config.db.get().await?;
    use crate::db::jobs::dsl as jobs;
    use crate::db::repositories::dsl as repos;
    use crate::db::users::dsl as users;

pub async fn ws_handler(

    if let Some(repo) = repos::repositories
        .inner_join(jobs::jobs)
        .inner_join(users::users)
        .filter(users::login.eq(&path.owner))
        .filter(repos::name.eq(&path.repo))
        .filter(repos::owner.nullable().eq(uid).or(crate::has_permissions!(
            uid.unwrap_or(uuid::Uuid::nil()),
            repos::id,
            Perm::WRITE_JOBS.bits()
        )))
        .filter(jobs::id.eq(path.job_id))
        .select(repos::id)
        .get_result::<uuid::Uuid>(&mut db)
        .await
        .optional()?
    {
        let mut db = config.db.get().await.unwrap();
        use crate::db::jobs::dsl as jobs;
        let (Some(state), channel) = jobs::jobs
            .find(&path.job_id)
            .select((jobs::repo_state, jobs::channel))
            .get_result::<(Option<String>, String)>(&mut db)
            .await
            .unwrap()
        else {
            return Err(crate::Error::NotFound);
        };
        let id = diesel::insert_into(jobs::jobs)
            .values((jobs::repo.eq(repo), jobs::repo_state.eq(&state)))
            .returning(jobs::id)
            .get_result::<uuid::Uuid>(&mut db)
            .await
            .unwrap();
        use crate::db::repositories::dsl as repositories;
        use crate::db::users::dsl as users;
        if let Some((state, deployment)) =
            crate::replication::get_config_state(&config.repo_locks, repo, channel)
                .await
                .unwrap()
        {
            let (owner, repo) = repositories::repositories
                .find(repo)
                .inner_join(users::users)
                .select((users::login, repositories::name))
                .get_result::<(String, String)>(&mut db)
                .await
                .unwrap();
            let mut targets = std::collections::HashMap::new();
            targets.insert(deployment.clone(), format!(".#{}", deployment));
            let body = serde_json::to_string(&ci::Trigger {
                id,
                owner,
                repo,
                state: state.to_base32(),
                targets,
            })
            .unwrap();
            debug!("{:?} {}", config.ci.url, body);
            tokio::spawn(async move {
                let client = reqwest::Client::new();
                for (n, ci_url) in config.ci.url.iter().enumerate() {
                    let res = client
                        .post(ci_url.clone() + "/trigger")
                        .header("Content-Type", "application/json")
                        .body(body.clone())
                        .send()
                        .await
                        .unwrap();
                    if let reqwest::StatusCode::OK = res.status() {
                        crate::replication::sync_job(&mut db, client, &config.ci, n, id).await;
                        break;
                    }
                }
            });
        }
        Ok(Redirect::to(&format!("/{}/{}/jobs", path.owner, path.repo)))
    } else {
        Err(crate::Error::NotFound)
    }
}
pub async fn kill(
    State(config): State<Config>,
    jar: SignedCookieJar,
    Path(path): Path<JobPath>,
) -> Result<Redirect, crate::Error> {
    let (uid, login) = if let Some((a, b)) = get_user_login(&jar, &config).await? {
        (Some(a), Some(b))
    } else {
        (None, None)
    };
    let mut db = config.db.get().await?;
    use crate::db::jobs::dsl as jobs;
    use crate::db::repositories::dsl as repos;
    use crate::db::users::dsl as users;
    if let Some(repo) = repos::repositories
        .inner_join(jobs::jobs)
        .inner_join(users::users)
        .filter(users::login.eq(&path.owner))
        .filter(repos::name.eq(&path.repo))
        .filter(repos::owner.nullable().eq(uid).or(crate::has_permissions!(
            uid.unwrap_or(uuid::Uuid::nil()),
            repos::id,
            Perm::WRITE_JOBS.bits()
        )))
        .filter(jobs::id.eq(path.job_id))
        .select(repos::id)
        .get_result::<uuid::Uuid>(&mut db)
        .await
        .optional()?
    {
        let client = reqwest::Client::new();
        for cl in config.ci.url.iter() {
            let res = client
                .post(format!("{}/kill/{}", cl, path.job_id))
                .header("Content-Type", "application/json")
                .send()
                .await
                .unwrap();
        }
        Ok(Redirect::to(&format!("/{}/{}/jobs", path.owner, path.repo)))
    } else {
        Err(crate::Error::NotFound)
    }
}
pub async fn job_ws_handler(

    jar: SignedCookieJar,

    ws.on_upgrade(move |socket| handle_socket(config, path.job_id, socket))

    use crate::db::repositories::dsl as repos;
    use crate::db::users::dsl as users;
    let (uid, login) = if let Ok(Some((a, b))) = get_user_login(&jar, &config).await {
        (Some(a), Some(b))
    } else {
        (None, None)
    };
    let mut db = config.db.get().await.unwrap();
    if repos::repositories
        .inner_join(users::users)
        .filter(users::login.eq(path.owner))
        .filter(repos::name.eq(path.repo))
        .filter(repos::owner.nullable().eq(uid).or(crate::has_permissions!(
            uid.unwrap_or(uuid::Uuid::nil()),
            repos::id,
            Perm::READ_JOBS.bits()
        )))
        .select(repos::id)
        .get_result::<uuid::Uuid>(&mut db)
        .await
        .optional()
        .unwrap()
        .is_none()
    {
        return crate::Error::NeedsAuth.into_response();
    }
    ws.on_upgrade(move |socket| job_handle_socket(config, path.job_id, socket))
}
pub async fn jobs_ws_handler(
    State(config): State<Config>,
    Path(path): Path<JobPath>,
    jar: SignedCookieJar,
    ws: WebSocketUpgrade,
) -> Response {
    use crate::db::repositories::dsl as repos;
    use crate::db::users::dsl as users;
    let (uid, login) = if let Some((a, b)) = get_user_login(&jar, &config).await.unwrap() {
        (Some(a), Some(b))
    } else {
        (None, None)
    };
    let mut db = config.db.get().await.unwrap();
    let Some(repo_id) = repos::repositories
        .inner_join(users::users)
        .filter(users::login.eq(path.owner))
        .filter(repos::name.eq(path.repo))
        .filter(repos::owner.nullable().eq(uid).or(crate::has_permissions!(
            uid.unwrap_or(uuid::Uuid::nil()),
            repos::id,
            Perm::READ_JOBS.bits()
        )))
        .select(repos::id)
        .get_result::<uuid::Uuid>(&mut db)
        .await
        .optional()
        .unwrap()
    else {
        return crate::Error::NeedsAuth.into_response();
    };
    ws.on_upgrade(move |socket| jobs_handle_socket(config, repo_id, socket))

enum Msg<'a> {

enum JobMsg<'a> {

async fn handle_socket(config: Config, id: uuid::Uuid, mut socket: WebSocket) {

async fn job_handle_socket(config: Config, id: uuid::Uuid, mut socket: WebSocket) {

        socket
            .send(
                serde_json::to_string(&Msg::Status {
                    ended: chrono::DateTime::UNIX_EPOCH,
                    status: None,
                })
                .unwrap()
                .into(),
            )
            .await
            .unwrap_or(());

                socket.send(serde_json::to_string(&Msg::Heartbeat).unwrap().into()).await.unwrap_or(());

                socket.send(serde_json::to_string(&JobMsg::Heartbeat).unwrap().into()).await.unwrap_or(());

                    socket.send(serde_json::to_string(&Msg::Status {

                    socket.send(serde_json::to_string(&JobMsg::Status {

                socket.send(serde_json::to_string(&Msg::Status {

                socket.send(serde_json::to_string(&JobMsg::Status {

                    serde_json::to_string(&Msg::Status { ended, status })

                    serde_json::to_string(&JobMsg::Status { ended, status })

                        serde_json::to_string(&Msg::Chunk {

                        serde_json::to_string(&JobMsg::Chunk {

                        serde_json::to_string(&Msg::Chunk {

                        serde_json::to_string(&JobMsg::Chunk {

            serde_json::to_string(&Msg::Chunk {

            serde_json::to_string(&JobMsg::Chunk {

            serde_json::to_string(&Msg::Chunk {

            serde_json::to_string(&JobMsg::Chunk {

    }
}
#[derive(Debug, Deserialize, Serialize)]
enum JobsMsg {
    Jobs(Vec<Job>),
    Heartbeat,
}
// Update jobs page list.
async fn jobs_handle_socket(config: Config, repo_id: uuid::Uuid, mut socket: WebSocket) {
    use crate::db::jobs::dsl as jobs;
    let mut latest = {
        let mut latest = config.jobs_latest.lock().await;
        match latest.entry(repo_id) {
            Entry::Vacant(e) => {
                let last = jobs::jobs
                    .filter(jobs::repo.eq(repo_id))
                    .select(jobs::id)
                    .order_by(jobs::started.desc())
                    .first::<uuid::Uuid>(&mut config.db.get().await.unwrap())
                    .await
                    .unwrap();
                let (sender, receiver) = tokio::sync::watch::channel(last);
                e.insert(sender);
                receiver
            }
            Entry::Occupied(e) => e.get().subscribe(),
        }
    };
    let mut interval = tokio::time::interval(std::time::Duration::from_secs(30));
    loop {
        debug!("waiting job");
        tokio::select! {
            _ = interval.tick() => {
                socket.send(serde_json::to_string(&JobsMsg::Heartbeat).unwrap().into()).await.unwrap_or(());
            }
            x = latest.changed() => {
                debug!("status {:?}", x);
                let mut db = config.db.get().await.unwrap();
                let latest = latest.borrow_and_update().clone();
                let jobs = jobs::jobs
                    .filter(jobs::repo.eq(repo_id))
                    .select(Job::as_select())
                    .order_by(jobs::started.desc())
                    .get_results::<Job>(&mut db)
                    .await.unwrap();
                socket.send(serde_json::to_string(&JobsMsg::Jobs(jobs)).unwrap().into()).await.unwrap_or(());
                break
            }
        }

use crate::proxy::HSTS;

use hyper::header::*;

use hyper::header::*;