Coverage / crates/hub/src/web/flow.rs 80.1%

pub fn why_not_run(enabled: bool, signed_in: bool, role: Role) -> Option<String> {

    if !enabled {

        Some("Flow is off for this repository; a site administrator can turn it on in Settings › Pipelines.".into())

    } else if !signed_in {

        Some("Sign in with write access to run the pipeline.".into())

    } else if role < Role::Write {

        None

}

pub async fn deployments(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name)): Path<(String, String)>,

) -> Result<Page<DeploymentsPage>, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/deployments"), &viewer);

    let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;

    let history = fdb::deployments(&app.db, row.id, 50).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    let mut current: Vec<fdb::Deployment> = Vec::new();

    for d in &history {

        if !current.iter().any(|c| c.environment == d.environment) {

            current.push(d.clone());

        }

    current.sort_by(|a, b| a.environment.cmp(&b.environment));

    Ok(Page(DeploymentsPage { shell, repo: RepoShell::new(&app, &row, "flow").with_role(role), current, history }))

}

pub async fn runs(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name)): Path<(String, String)>,

    Query(q): Query<RunsQuery>,

) -> Result<Page<RunsPage>, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow"), &viewer);

    let (_repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;

    let list = fdb::runs(&app.db, row.id, PER_PAGE + 1, q.before).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    let older = (list.len() as i64 > PER_PAGE).then(|| list[PER_PAGE as usize - 1].id);

    let queue = if list.iter().any(|r| r.live()) { fdb::queue(&app.db).await.unwrap_or_default() } else { vec![] };

    let mut runs = Vec::new();

    for r in list.into_iter().take(PER_PAGE as usize) {

        let place = if r.live() { crate::flow::queue_info(&queue, row.id, r.id).place() } else { None };

        let jobs = fdb::jobs(&app.db, r.id).await.unwrap_or_default();

        let duration = match r.finished_ms {

            Some(f) => crate::flow::human_ms(f - r.created_ms),

            None if r.live() => crate::flow::human_ms(db::now_ms() - r.created_ms),

        runs.push(RunRow { jobs: jobs.into_iter().map(|j| (j.name, j.status)).collect(), duration, r, place });

    let enabled = crate::flow::enabled(&app, &row);

    let why_not = why_not_run(enabled, viewer.account().is_some(), role);

    let can_run = why_not.is_none();

    let bookmarks = if can_run { bookmark_names(&app, &row).await } else { vec![] };

    Ok(Page(RunsPage {

        shell,

        repo: RepoShell::new(&app, &row, "flow").with_role(role),

        runs,

        enabled,

        older,

        can_run,

        bookmarks,

        error: q.error,

        why_not,

    }))

}

pub fn wasm_jobs(p: &flow_core::Pipeline) -> Vec<TabJob> {

    p.jobs

        .values()

        .filter_map(|j| {

            let sb = p.sandboxes.get(&j.sandbox)?;

            (sb.runtime == flow_core::Runtime::Wasm && !j.exec.is_empty()).then(|| {

                let mut env = sb.env.clone();

                env.extend(j.env.clone());

                TabJob { name: j.name.clone(), exec: j.exec.clone(), env, memory_mib: sb.memory_mib }

            })

        })

        .collect()

}

pub async fn tab(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name)): Path<(String, String)>,

    Query(q): Query<TabQuery>,

) -> Result<Page<TabPage>, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/tab"), &viewer);

    let (repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;

    let bookmarks = bookmark_names(&app, &row).await;

    let bookmark = q.bookmark.filter(|b| bookmarks.contains(b)).unwrap_or_else(|| row.default_bookmark.clone());

    let b = bookmark.clone();

    let text = app

        .repos

        .read(&repo, move |r, cache| {

            let Some(st) = r.view()?.bookmarks.get(&b).copied() else { return Ok(None) };

            let heads = r.store().get_state(&st)?.heads;

            let loaded = cache.load(r, &heads)?;

            Ok(match loaded.snapshot.files.get(flow_core::PIPELINE_PATH).map(|f| &f.content) {

                Some(oxy_core::state::FileContent::Text { bytes, .. }) => Some(String::from_utf8_lossy(bytes).into_owned()),

        })

        .await

        .unwrap_or(None);

    let (jobs, problem) = match text.map(|t| flow_core::Pipeline::parse(&t)) {

        Some(Ok(p)) => {

            let jobs = wasm_jobs(&p);

            let problem =

                jobs.is_empty().then(|| format!("The pipeline on {bookmark} has no WASM jobs (a sandbox with runtime \"wasm\")."));

            (jobs, problem)

        shell,

        repo: RepoShell::new(&app, &row, "flow").with_role(role),

        archive: format!("/{owner}/{name}/archive?ref={}&format=zip", urlencoding(&bookmark)),

        bookmark,

        bookmarks,

        jobs: jobs.len(),

        jobs_json: serde_json::to_string(&jobs).unwrap_or_else(|_| "[]".into()),

        problem,

}

async fn bookmark_names(app: &Shared, row: &db::repos::RepoRow) -> Vec<String> {

    let Some(repo) = crate::repos::RepoName::new(&row.owner, &row.name) else { return vec![] };

    let mut names: Vec<String> = app.repos.read(&repo, |r, _| Ok(r.view()?.bookmarks.keys().cloned().collect())).await.unwrap_or_default();

    names.sort_by_key(|n| (n != &row.default_bookmark, n.clone()));

    names

}

pub async fn expedite(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name, number)): Path<(String, String, i64)>,

    Form(f): Form<ExpediteForm>,

) -> Result<Response, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/{number}"), &viewer);

    let me = require(&viewer, &shell)?;

    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Maintain).await?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    let run = fdb::run(&app.db, row.id, number)

        .await

        .map_err(|e| Error::from_db(shell.clone(), e))?

        .ok_or_else(|| Error::not_found(shell.clone()))?;

    if let Err(msg) = crate::flow::expedite(&app, &row, &run, &me.account, &f.reason).await {

    }

    Ok(Redirect::to(&format!("/{owner}/{name}/flow/{number}")).into_response())

}

fn urlencoding(s: &str) -> String {

    s.bytes()

        .map(|b| match b {

            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => (b as char).to_string(),

        })

        .collect()

}

pub fn layout(jobs: Vec<fdb::Job>) -> Graph {

    let mut by_depth: std::collections::BTreeMap<i64, Vec<fdb::Job>> = Default::default();

    for j in jobs {

        by_depth.entry(j.depth).or_default().push(j);

    }

    let rows = by_depth.values().map(Vec::len).max().unwrap_or(0) as i64;

    let height = PAD * 2 + rows * NODE_H + (rows - 1).max(0) * ROW_GAP;

    let mut nodes = Vec::new();

    for (col, (_, mut js)) in by_depth.into_iter().enumerate() {

        js.sort_by(|a, b| a.name.cmp(&b.name));

        let n = js.len() as i64;

        let column_h = n * NODE_H + (n - 1).max(0) * ROW_GAP;

        let top = (height - column_h) / 2;

        for (i, j) in js.into_iter().enumerate() {

            let duration = crate::flow::duration(&j);

            nodes.push(Node { x: PAD + col as i64 * (NODE_W + COL_GAP), y: top + i as i64 * (NODE_H + ROW_GAP), duration, j });

        }