diff --git a/src/api/jobs.test.ts b/src/api/jobs.test.ts new file mode 100644 index 0000000..01df409 --- /dev/null +++ b/src/api/jobs.test.ts @@ -0,0 +1,240 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { + listJobs, + getJob, + cancelJob, + JobApiError, + type JobApiConfig, + type Job, +} from "./jobs.js"; + +// Mock fetch globally +const mockFetch = vi.fn(); +global.fetch = mockFetch; + +function expectFromParam(url: string) { + const parsed = new URL(url); + expect(parsed.searchParams.get("from")).toBe("ts-sdk"); + return parsed; +} + +describe("Job API client", () => { + const config: JobApiConfig = { + baseUrl: "https://api.tinybird.co", + token: "p.admin-token", + }; + + const sampleJob: Job = { + id: "job-123", + kind: "import", + status: "done", + created_at: "2024-01-01T00:00:00.000Z", + updated_at: "2024-01-01T00:01:00.000Z", + job_url: "https://api.tinybird.co/v0/jobs/job-123", + }; + + beforeEach(() => { + mockFetch.mockReset(); + }); + + describe("listJobs", () => { + it("lists jobs with no filters", async () => { + mockFetch.mockResolvedValueOnce({ + ok: true, + json: () => Promise.resolve({ jobs: [sampleJob] }), + }); + + const result = await listJobs(config); + + expect(result).toEqual([sampleJob]); + + const [url, init] = mockFetch.mock.calls[0]; + const parsed = expectFromParam(url); + expect(parsed.pathname).toBe("/v0/jobs"); + expect(parsed.searchParams.get("status")).toBeNull(); + expect(parsed.searchParams.get("kind")).toBeNull(); + expect(init.method).toBe("GET"); + const headers = new Headers(init.headers); + expect(headers.get("Authorization")).toBe("Bearer p.admin-token"); + }); + + it("filters by status", async () => { + mockFetch.mockResolvedValueOnce({ + ok: true, + json: () => Promise.resolve({ jobs: [] }), + }); + + await listJobs(config, { status: "error" }); + + const [url] = mockFetch.mock.calls[0]; + const parsed = new URL(url); + expect(parsed.searchParams.get("status")).toBe("error"); + }); + + it("filters by kind", async () => { + mockFetch.mockResolvedValueOnce({ + ok: true, + json: () => Promise.resolve({ jobs: [] }), + }); + + await listJobs(config, { kind: "populateview" }); + + const [url] = mockFetch.mock.calls[0]; + const parsed = new URL(url); + expect(parsed.searchParams.get("kind")).toBe("populateview"); + }); + + it("filters by status and kind together", async () => { + mockFetch.mockResolvedValueOnce({ + ok: true, + json: () => Promise.resolve({ jobs: [] }), + }); + + await listJobs(config, { status: "working", kind: "copy" }); + + const [url] = mockFetch.mock.calls[0]; + const parsed = new URL(url); + expect(parsed.searchParams.get("status")).toBe("working"); + expect(parsed.searchParams.get("kind")).toBe("copy"); + }); + + it("returns an empty array when the API omits the jobs field", async () => { + mockFetch.mockResolvedValueOnce({ + ok: true, + json: () => Promise.resolve({}), + }); + + const result = await listJobs(config); + expect(result).toEqual([]); + }); + + it("uses custom fetch implementation when provided", async () => { + const customFetch = vi.fn().mockResolvedValueOnce({ + ok: true, + json: () => Promise.resolve({ jobs: [] }), + }); + + await listJobs({ ...config, fetch: customFetch as typeof fetch }); + + expect(customFetch).toHaveBeenCalledTimes(1); + expect(mockFetch).not.toHaveBeenCalled(); + }); + + it("throws JobApiError on failure", async () => { + mockFetch.mockResolvedValueOnce({ + ok: false, + status: 500, + statusText: "Internal Server Error", + text: () => Promise.resolve("boom"), + }); + + await expect(listJobs(config)).rejects.toThrow(JobApiError); + }); + }); + + describe("getJob", () => { + it("gets a job by id", async () => { + mockFetch.mockResolvedValueOnce({ + ok: true, + json: () => Promise.resolve(sampleJob), + }); + + const result = await getJob(config, "job-123"); + expect(result).toEqual(sampleJob); + + const [url, init] = mockFetch.mock.calls[0]; + const parsed = expectFromParam(url); + expect(parsed.pathname).toBe("/v0/jobs/job-123"); + expect(init.method).toBe("GET"); + const headers = new Headers(init.headers); + expect(headers.get("Authorization")).toBe("Bearer p.admin-token"); + }); + + it("encodes the job id in the path", async () => { + mockFetch.mockResolvedValueOnce({ + ok: true, + json: () => Promise.resolve(sampleJob), + }); + + await getJob(config, "job/with/slash"); + + const [url] = mockFetch.mock.calls[0]; + const parsed = new URL(url); + expect(parsed.pathname).toBe("/v0/jobs/job%2Fwith%2Fslash"); + }); + + it("throws a helpful JobApiError on 404", async () => { + mockFetch.mockResolvedValueOnce({ + ok: false, + status: 404, + statusText: "Not Found", + text: () => Promise.resolve(""), + }); + + await expect(getJob(config, "missing-job")).rejects.toThrow(JobApiError); + + mockFetch.mockResolvedValueOnce({ + ok: false, + status: 404, + statusText: "Not Found", + text: () => Promise.resolve(""), + }); + + try { + await getJob(config, "missing-job"); + } catch (error) { + expect((error as JobApiError).status).toBe(404); + expect((error as JobApiError).message).toContain("missing-job"); + expect((error as JobApiError).message).toContain("not found"); + } + }); + }); + + describe("cancelJob", () => { + it("cancels a job", async () => { + const cancellingJob: Job = { ...sampleJob, status: "cancelling" }; + mockFetch.mockResolvedValueOnce({ + ok: true, + json: () => Promise.resolve(cancellingJob), + }); + + const result = await cancelJob(config, "job-123"); + expect(result).toEqual(cancellingJob); + + const [url, init] = mockFetch.mock.calls[0]; + const parsed = expectFromParam(url); + expect(parsed.pathname).toBe("/v0/jobs/job-123/cancel"); + expect(init.method).toBe("POST"); + const headers = new Headers(init.headers); + expect(headers.get("Authorization")).toBe("Bearer p.admin-token"); + }); + + it("throws a helpful JobApiError on 404", async () => { + mockFetch.mockResolvedValueOnce({ + ok: false, + status: 404, + statusText: "Not Found", + text: () => Promise.resolve(""), + }); + + await expect(cancelJob(config, "missing-job")).rejects.toThrow(JobApiError); + }); + + it("throws JobApiError with response body on other failures", async () => { + mockFetch.mockResolvedValueOnce({ + ok: false, + status: 409, + statusText: "Conflict", + text: () => Promise.resolve('{"error": "Job already finished"}'), + }); + + try { + await cancelJob(config, "job-123"); + expect.unreachable(); + } catch (error) { + expect((error as JobApiError).status).toBe(409); + expect((error as JobApiError).message).toContain("Job already finished"); + } + }); + }); +}); diff --git a/src/api/jobs.ts b/src/api/jobs.ts new file mode 100644 index 0000000..3ed1e70 --- /dev/null +++ b/src/api/jobs.ts @@ -0,0 +1,246 @@ +/** + * Tinybird Job management API client + * Wraps the /v0/jobs endpoints used to inspect and cancel async jobs + * (ingest/append/replace operations, connector syncs, populates, etc.) + * + * Note: there is no confirmed `/v0/jobs/{id}/retry` endpoint, so this module + * intentionally does not expose a `retry` operation. See PROTM-2270. + */ + +import { createTinybirdFetcher } from "./fetcher.js"; + +/** + * API configuration for job operations + */ +export interface JobApiConfig { + /** Tinybird API base URL */ + baseUrl: string; + /** Workspace or branch token used to authenticate requests */ + token: string; + /** Custom fetch implementation (optional, defaults to global fetch) */ + fetch?: typeof fetch; + /** Default timeout in milliseconds applied to every request (optional) */ + timeout?: number; +} + +/** + * Per-request options shared by job operations + */ +export interface JobRequestOptions { + /** Request timeout in milliseconds (overrides `JobApiConfig.timeout`) */ + timeout?: number; + /** AbortController signal for cancellation */ + signal?: AbortSignal; +} + +/** + * Status values reported for a Tinybird job + */ +export type JobStatus = + | "waiting" + | "working" + | "done" + | "error" + | "cancelling" + | "cancelled"; + +/** + * A Tinybird job as returned by the Jobs API. + * + * The exact set of fields present depends on the job `kind` (e.g. `import`, + * `populateview`, `copy`, `datasource_create_from_connector`, ...), so + * additional kind-specific fields are preserved via the index signature. + */ +export interface Job { + /** Job identifier */ + id: string; + /** Job kind (e.g. "import", "populateview", "copy") */ + kind: string; + /** Current job status */ + status: JobStatus; + /** ISO timestamp of when the job was created */ + created_at: string; + /** ISO timestamp of the last update to the job */ + updated_at: string; + /** URL to fetch this job's status */ + job_url?: string; + /** Error message, present when `status` is "error" */ + error?: string | null; + /** Additional kind-specific fields returned by the API */ + [key: string]: unknown; +} + +/** + * Options for filtering the jobs list + */ +export interface ListJobsOptions extends JobRequestOptions { + /** Only return jobs with this status */ + status?: JobStatus; + /** Only return jobs of this kind */ + kind?: string; +} + +/** + * Error thrown by job API operations + */ +export class JobApiError extends Error { + constructor( + message: string, + public readonly status: number, + public readonly body?: unknown + ) { + super(message); + this.name = "JobApiError"; + } +} + +function getFetch(config: JobApiConfig) { + return createTinybirdFetcher(config.fetch ?? globalThis.fetch); +} + +function createAbortSignal( + config: JobApiConfig, + options?: JobRequestOptions +): AbortSignal | undefined { + const timeoutMs = options?.timeout ?? config.timeout; + const existingSignal = options?.signal; + + if (!timeoutMs) { + return existingSignal; + } + + const timeoutSignal = AbortSignal.timeout(timeoutMs); + + if (!existingSignal) { + return timeoutSignal; + } + + return AbortSignal.any([timeoutSignal, existingSignal]); +} + +async function handleErrorResponse( + response: Response, + notFoundMessage?: string +): Promise { + const body = await response.text(); + + let message: string; + if (response.status === 404 && notFoundMessage) { + message = notFoundMessage; + } else { + message = `Request failed: ${response.status} ${response.statusText}`; + if (body) { + message += `\nAPI response: ${body}`; + } + } + + throw new JobApiError(message, response.status, body || undefined); +} + +/** + * List jobs in the workspace, optionally filtered by status and/or kind. + * GET /v0/jobs + * + * @param config - API configuration + * @param options - Optional status/kind filters + * @returns Array of jobs + */ +export async function listJobs( + config: JobApiConfig, + options: ListJobsOptions = {} +): Promise { + const fetchFn = getFetch(config); + const url = new URL("/v0/jobs", `${config.baseUrl}/`); + + if (options.status) { + url.searchParams.set("status", options.status); + } + if (options.kind) { + url.searchParams.set("kind", options.kind); + } + + const response = await fetchFn(url.toString(), { + method: "GET", + headers: { + Authorization: `Bearer ${config.token}`, + }, + signal: createAbortSignal(config, options), + }); + + if (!response.ok) { + await handleErrorResponse(response); + } + + const data = (await response.json()) as { jobs?: Job[] }; + return data.jobs ?? []; +} + +/** + * Get details for a single job. + * GET /v0/jobs/{job_id} + * + * @param config - API configuration + * @param jobId - Job ID + * @param options - Request options + * @returns Job details + */ +export async function getJob( + config: JobApiConfig, + jobId: string, + options: JobRequestOptions = {} +): Promise { + const fetchFn = getFetch(config); + const url = new URL(`/v0/jobs/${encodeURIComponent(jobId)}`, `${config.baseUrl}/`); + + const response = await fetchFn(url.toString(), { + method: "GET", + headers: { + Authorization: `Bearer ${config.token}`, + }, + signal: createAbortSignal(config, options), + }); + + if (!response.ok) { + await handleErrorResponse(response, `Job '${jobId}' not found.`); + } + + return (await response.json()) as Job; +} + +/** + * Cancel a running job. + * POST /v0/jobs/{job_id}/cancel + * + * Note: not all job kinds/states support cancellation; the API may respond + * with the job already transitioning to "cancelling" rather than "cancelled". + * + * @param config - API configuration + * @param jobId - Job ID + * @param options - Request options + * @returns The updated job + */ +export async function cancelJob( + config: JobApiConfig, + jobId: string, + options: JobRequestOptions = {} +): Promise { + const fetchFn = getFetch(config); + const url = new URL( + `/v0/jobs/${encodeURIComponent(jobId)}/cancel`, + `${config.baseUrl}/` + ); + + const response = await fetchFn(url.toString(), { + method: "POST", + headers: { + Authorization: `Bearer ${config.token}`, + }, + signal: createAbortSignal(config, options), + }); + + if (!response.ok) { + await handleErrorResponse(response, `Job '${jobId}' not found.`); + } + + return (await response.json()) as Job; +} diff --git a/src/client/base.ts b/src/client/base.ts index 637c0c8..600e35f 100644 --- a/src/client/base.ts +++ b/src/client/base.ts @@ -21,6 +21,7 @@ import { TinybirdError } from "./types.js"; import { TinybirdApi, TinybirdApiError } from "../api/api.js"; import type { CreateBranchOptions } from "../api/branches.js"; import { TokensNamespace } from "./tokens.js"; +import { JobsNamespace } from "./jobs.js"; /** * Resolved token info from dev mode @@ -74,6 +75,9 @@ export class TinybirdClient { /** Token operations (JWT creation, etc.) */ public readonly tokens: TokensNamespace; + /** Job management operations (list, get, cancel) */ + public readonly jobs: JobsNamespace; + constructor(config: ClientConfig) { // Validate required config if (!config.baseUrl) { @@ -122,6 +126,14 @@ export class TinybirdClient { this.config.fetch, this.config.timeout ); + + // Initialize jobs namespace + this.jobs = new JobsNamespace( + () => this.getToken(), + this.config.baseUrl, + this.config.fetch, + this.config.timeout + ); } /** diff --git a/src/client/jobs.test.ts b/src/client/jobs.test.ts new file mode 100644 index 0000000..bcfbf59 --- /dev/null +++ b/src/client/jobs.test.ts @@ -0,0 +1,137 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { TinybirdClient } from "./base.js"; +import { TinybirdError } from "./types.js"; + +const mockFetch = vi.fn(); +global.fetch = mockFetch; + +describe("JobsNamespace", () => { + beforeEach(() => { + mockFetch.mockReset(); + }); + + it("uses client custom fetch for list", async () => { + const customFetch = vi.fn().mockResolvedValueOnce({ + ok: true, + json: () => + Promise.resolve({ + jobs: [ + { + id: "job-1", + kind: "import", + status: "done", + created_at: "2024-01-01T00:00:00.000Z", + updated_at: "2024-01-01T00:01:00.000Z", + }, + ], + }), + }); + + const client = new TinybirdClient({ + baseUrl: "https://api.tinybird.co", + token: "p.admin-token", + fetch: customFetch as typeof fetch, + }); + + const result = await client.jobs.list({ status: "done" }); + + expect(result).toHaveLength(1); + expect(result[0].id).toBe("job-1"); + expect(customFetch).toHaveBeenCalledTimes(1); + expect(mockFetch).not.toHaveBeenCalled(); + + const [url, init] = customFetch.mock.calls[0] as [string, RequestInit]; + const parsed = new URL(url); + expect(parsed.pathname).toBe("/v0/jobs"); + expect(parsed.searchParams.get("status")).toBe("done"); + expect(parsed.searchParams.get("from")).toBe("ts-sdk"); + expect(init.method).toBe("GET"); + const headers = new Headers(init.headers); + expect(headers.get("Authorization")).toBe("Bearer p.admin-token"); + }); + + it("gets a single job", async () => { + const customFetch = vi.fn().mockResolvedValueOnce({ + ok: true, + json: () => + Promise.resolve({ + id: "job-123", + kind: "populateview", + status: "working", + created_at: "2024-01-01T00:00:00.000Z", + updated_at: "2024-01-01T00:01:00.000Z", + }), + }); + + const client = new TinybirdClient({ + baseUrl: "https://api.tinybird.co", + token: "p.admin-token", + fetch: customFetch as typeof fetch, + }); + + const job = await client.jobs.get("job-123"); + + expect(job.status).toBe("working"); + + const [url, init] = customFetch.mock.calls[0] as [string, RequestInit]; + const parsed = new URL(url); + expect(parsed.pathname).toBe("/v0/jobs/job-123"); + expect(init.method).toBe("GET"); + }); + + it("cancels a job", async () => { + const customFetch = vi.fn().mockResolvedValueOnce({ + ok: true, + json: () => + Promise.resolve({ + id: "job-123", + kind: "import", + status: "cancelling", + created_at: "2024-01-01T00:00:00.000Z", + updated_at: "2024-01-01T00:01:00.000Z", + }), + }); + + const client = new TinybirdClient({ + baseUrl: "https://api.tinybird.co", + token: "p.admin-token", + fetch: customFetch as typeof fetch, + }); + + const job = await client.jobs.cancel("job-123"); + + expect(job.status).toBe("cancelling"); + + const [url, init] = customFetch.mock.calls[0] as [string, RequestInit]; + const parsed = new URL(url); + expect(parsed.pathname).toBe("/v0/jobs/job-123/cancel"); + expect(init.method).toBe("POST"); + }); + + it("wraps job API errors as TinybirdError", async () => { + const errorResponse = { + ok: false, + status: 404, + statusText: "Not Found", + text: () => Promise.resolve(""), + }; + const customFetch = vi + .fn() + .mockResolvedValueOnce(errorResponse) + .mockResolvedValueOnce(errorResponse); + + const client = new TinybirdClient({ + baseUrl: "https://api.tinybird.co", + token: "p.admin-token", + fetch: customFetch as typeof fetch, + }); + + await expect(client.jobs.get("missing-job")).rejects.toThrow(TinybirdError); + + try { + await client.jobs.get("missing-job"); + } catch (error) { + expect((error as TinybirdError).statusCode).toBe(404); + } + }); +}); diff --git a/src/client/jobs.ts b/src/client/jobs.ts new file mode 100644 index 0000000..4e6594f --- /dev/null +++ b/src/client/jobs.ts @@ -0,0 +1,148 @@ +/** + * Job management namespace for TinybirdClient + * + * Wraps `/v0/jobs` so callers can inspect and manage the async jobs the SDK + * (or `tb`/the UI) triggers, such as ingest/append/replace operations and + * connector syncs. + * + * Note: a dedicated job "retry" endpoint could not be confirmed against the + * Tinybird backend, so this namespace intentionally only exposes + * `list`, `get` and `cancel`. See PROTM-2270. + */ + +import type { + Job, + JobRequestOptions, + ListJobsOptions, +} from "../api/jobs.js"; +import { + listJobs as apiListJobs, + getJob as apiGetJob, + cancelJob as apiCancelJob, + JobApiError, +} from "../api/jobs.js"; +import { TinybirdError } from "./types.js"; + +/** + * Job management namespace for TinybirdClient + */ +export class JobsNamespace { + constructor( + private readonly getToken: () => Promise, + private readonly baseUrl: string, + private readonly fetchFn?: typeof globalThis.fetch, + private readonly timeout?: number + ) {} + + /** + * List jobs in the workspace, optionally filtered by status and/or kind. + * + * @param options - Optional status/kind filters + * @returns Array of jobs + * + * @example + * ```ts + * const jobs = await client.jobs.list(); + * + * const failedImports = await client.jobs.list({ + * status: "error", + * kind: "import", + * }); + * ``` + */ + async list(options: ListJobsOptions = {}): Promise { + const token = await this.getToken(); + + try { + return await apiListJobs( + { + baseUrl: this.baseUrl, + token, + fetch: this.fetchFn, + timeout: this.timeout, + }, + options + ); + } catch (error) { + this.rethrow(error); + } + } + + /** + * Get details for a single job. + * + * @param jobId - Job ID + * @param options - Request options + * @returns Job details + * + * @example + * ```ts + * const job = await client.jobs.get("job_abc123"); + * console.log(job.status); // "done" + * ``` + */ + async get(jobId: string, options: JobRequestOptions = {}): Promise { + const token = await this.getToken(); + + try { + return await apiGetJob( + { + baseUrl: this.baseUrl, + token, + fetch: this.fetchFn, + timeout: this.timeout, + }, + jobId, + options + ); + } catch (error) { + this.rethrow(error); + } + } + + /** + * Cancel a running job. + * + * Not all job kinds/states support cancellation; the API may respond with + * the job transitioning to "cancelling" rather than immediately + * "cancelled". + * + * @param jobId - Job ID + * @param options - Request options + * @returns The updated job + * + * @example + * ```ts + * const job = await client.jobs.cancel("job_abc123"); + * console.log(job.status); // "cancelling" or "cancelled" + * ``` + */ + async cancel(jobId: string, options: JobRequestOptions = {}): Promise { + const token = await this.getToken(); + + try { + return await apiCancelJob( + { + baseUrl: this.baseUrl, + token, + fetch: this.fetchFn, + timeout: this.timeout, + }, + jobId, + options + ); + } catch (error) { + this.rethrow(error); + } + } + + private rethrow(error: unknown): never { + if (error instanceof JobApiError) { + throw new TinybirdError(error.message, error.status, { + error: error.message, + status: error.status, + }); + } + throw error; + } +} diff --git a/src/index.ts b/src/index.ts index b8bab18..84e002e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -315,6 +315,16 @@ export type { CreateJWTResult, } from "./api/tokens.js"; +// ============ Job API ============ +export { listJobs, getJob, cancelJob, JobApiError } from "./api/jobs.js"; +export type { + JobApiConfig, + JobRequestOptions, + JobStatus, + Job, + ListJobsOptions, +} from "./api/jobs.js"; + // ============ Config Types ============ // Import from config-types.ts to avoid bundling esbuild in client code export type { TinybirdConfig, DevMode, BranchDataMode } from "./cli/config-types.js";