New Completed Job Run

//native

import * as wmill from "windmill-client"

/**

 * New Completed Job Run

 * Emits job runs that have reached a terminal state since the last poll, tracked via Windmill state. Optionally restrict to a single job.

 */

export async function main(auth: RT.Databricks, job_id: number | undefined) {

  const base = auth.workspace_url.replace(/\/$/, "")

  const lastChecked: number = (await wmill.getState()) ?? 0

  const url = new URL(`${base}/api/2.2/jobs/runs/list`)

  url.searchParams.append("completed_only", "true")

  url.searchParams.append("limit", "25")

  if (job_id !== undefined) url.searchParams.append("job_id", String(job_id))

  const response = await fetch(url, {

    method: "GET",

    headers: {

      Authorization: `Bearer ${auth.token}`,

      Accept: "application/json",

    },

  })

  if (!response.ok) {

    throw new Error(`${response.status} ${await response.text()}`)

  }

  const data = await response.json()

  const runs: Array<{ end_time?: number; start_time?: number }> =

    data.runs ?? []

  const ts = (r: { end_time?: number; start_time?: number }) =>

    r.end_time || r.start_time || 0

  const maxTs = runs.reduce((m, r) => Math.max(m, ts(r)), 0)

  // First run: set the watermark to the newest completion seen and don't emit a backlog.

  if (!lastChecked) {

    await wmill.setState(maxTs)

    return []

  }

  const fresh = runs

    .filter((r) => ts(r) > lastChecked)

    .sort((a, b) => ts(a) - ts(b))

  if (maxTs > lastChecked) {

    await wmill.setState(maxTs)

  }

  return fresh

}