Write transactions begin IMMEDIATE, so a concurrent writer is waited for (busy timeout) instead of failing at once with database is locked; a test of 16 concurrent log appends fails without it 941abdd6bb4c

}

pub async fn create_org(pool: &Pool, name: &str, display_name: &str, owner: i64) -> sqlx::Result<Organisation> {

    let org: Organisation = sqlx::query_as("INSERT INTO organisations (name, display_name, created_ms) VALUES (?, ?, ?) RETURNING *")

        .bind(name)

        .bind(display_name)

}

pub async fn create_run(pool: &Pool, r: NewRun<'_>, jobs: &[NewJob]) -> sqlx::Result<i64> {

    let now = now_ms();

    let status = if r.error.is_some() { "error" } else { "queued" };

    let id: i64 = sqlx::query_scalar(

/// The oldest runnable job that fits in `room`.

pub async fn claim_within(pool: &Pool, runtimes: &[String], runner: &str, token_hash: &str, room: Room) -> sqlx::Result<Option<Job>> {

    let queued: Vec<Job> = sqlx::query_as(

        "SELECT j.* FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id WHERE j.status = 'queued' AND r.status IN ('queued', 'running') ORDER BY j.run_id, j.depth, j.id LIMIT 200",

    )

/// Append records; returns how many were kept (the rest exceed the cap).

pub async fn append_log(pool: &Pool, job_id: i64, records: &[flow_core::api::Record], cap: i64) -> sqlx::Result<usize> {

    let count: i64 = sqlx::query_scalar("SELECT log_count FROM flow_jobs WHERE id = ?").bind(job_id).fetch_one(&mut *tx).await?;

    let room = (cap - count).max(0) as usize;

    let keep = &records[..records.len().min(room)];

/// Record a job's end, skip the jobs that can no longer run, and settle the

/// run. Returns the jobs whose status changed (for their checks).

pub async fn finish(pool: &Pool, job_id: i64, o: Outcome<'_>) -> sqlx::Result<Vec<Job>> {

    let now = now_ms();

    let j: Job = sqlx::query_as("SELECT * FROM flow_jobs WHERE id = ?").bind(job_id).fetch_one(&mut *tx).await?;

    sqlx::query("UPDATE flow_jobs SET status = ?, exit_code = ?, error = ?, receipt = ?, receipt_hash = ?, finished_ms = ?, token_hash = '' WHERE id = ?")

pub fn now_ms() -> i64 {

    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as i64).unwrap_or(0)

}

pub const DEFAULT_COLUMNS: [&str; 3] = ["Backlog", "Doing", "Done"];

pub async fn create(pool: &Pool, owner: &str, title: &str, description: &str, public: bool) -> sqlx::Result<Project> {

    let number: i64 =

        sqlx::query_scalar("SELECT COALESCE(MAX(number), 0) + 1 FROM projects WHERE owner = ?").bind(owner).fetch_one(&mut *tx).await?;

    let now = now_ms();

    if !cols.iter().any(|c| c.id == column_id) {

        return Ok(false);

    }

    let base: i64 = sqlx::query_scalar("SELECT COALESCE(MAX(position), -1) + 1 FROM project_cards WHERE column_id = ?")

        .bind(to.id)

        .fetch_one(&mut *tx)

/// Take the next number of a repository's shared number space.

pub async fn next_number(pool: &Pool, repo_id: i64) -> sqlx::Result<i64> {

    sqlx::query("INSERT OR IGNORE INTO numbers (repo_id, next) VALUES (?, 1)").bind(repo_id).execute(&mut *tx).await?;

    let n: i64 = sqlx::query_scalar("SELECT next FROM numbers WHERE repo_id = ?").bind(repo_id).fetch_one(&mut *tx).await?;

    sqlx::query("UPDATE numbers SET next = next + 1 WHERE repo_id = ?").bind(repo_id).execute(&mut *tx).await?;

/// Record a push's per-reference results and bump the repository's activity.

pub async fn record_push(pool: &Pool, id: i64, pusher: Option<(i64, &str)>, results: &[PushOutcome]) -> sqlx::Result<()> {

    let now = now_ms();

    for r in results {

        sqlx::query("INSERT INTO pushes (repo_id, at_ms, reference, ok, reason, old_state, new_state, heads, changes, account_id, pusher) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")

            .bind(id)

    assert!(!read("vault"), "not pinned for this pusher");

    assert!(!read("other"), "public, but not the job's to read");

}