From dd868ecbf8046e0fa56bc47b27cb158afdfc86c9 Mon Sep 17 00:00:00 2001 From: Allen7702 Date: Sat, 26 Sep 2026 12:43:06 +0300 Subject: [PATCH 1/5] feat: implement Open Climate Service sync handler with configuration and job management --- src/services/worker/constants/handlers.ts | 19 +- .../worker/services/handlers/index.ts | 1 + .../handlers/openClimateServiceSync/index.ts | 235 +++++++++++++++++ .../openClimateServiceSync/schemas/config.ts | 24 ++ src/shared/clients/climateSync.ts | 161 ++++++++++++ src/tests/unit/handlers-catalog.test.ts | 9 +- .../handlers/openClimateServiceSync.test.ts | 237 ++++++++++++++++++ 7 files changed, 684 insertions(+), 2 deletions(-) create mode 100644 src/services/worker/services/handlers/openClimateServiceSync/index.ts create mode 100644 src/services/worker/services/handlers/openClimateServiceSync/schemas/config.ts create mode 100644 src/shared/clients/climateSync.ts create mode 100644 src/tests/unit/handlers/openClimateServiceSync.test.ts diff --git a/src/services/worker/constants/handlers.ts b/src/services/worker/constants/handlers.ts index d1ce8fa..3e41262 100644 --- a/src/services/worker/constants/handlers.ts +++ b/src/services/worker/constants/handlers.ts @@ -15,6 +15,8 @@ import { alertGenerationConfigSchema, alertGenerationContextSchema, } from "@/services/worker/services/handlers/alertGeneration/schemas/config.ts"; +import { openClimateServiceSyncConfigSchema } from "@/services/worker/services/handlers/openClimateServiceSync/schemas/config.ts"; +import { dhis2DataUploadConfigSchema } from "@/services/worker/services/handlers/dhis2DataUpload/schemas/config.ts"; import { dhis2AnalyticsRunConfigSchema } from "@/services/worker/services/handlers/dhis2AnalyticsRun/schemas/config.ts"; import type { ZodType } from "zod"; @@ -31,6 +33,7 @@ export type HandlerRegistryEntry = { }; export enum Handlers { + OPEN_CLIMATE_SERVICE_SYNC = "open-climate-service-sync", CLIMATE_OPENEO_CREATE = "climate-openeo-create", CLIMATE_OPENEO_POLL = "climate-openeo-poll", CLIMATE_OPENEO_DOWNLOAD = "climate-openeo-download", @@ -48,6 +51,18 @@ export const HANDLERS: Map = new Map< Handlers, HandlerRegistryEntry >([ + [ + Handlers.OPEN_CLIMATE_SERVICE_SYNC, + { + queueName: "step.open-climate-service-sync", + displayName: "Open Climate Service Sync", + description: "Bring Open Climate Service datasets up to date before the pipeline reads them", + tags: ["climate", "sync"], + schemas: { + config: openClimateServiceSyncConfigSchema, + }, + }, + ], [ Handlers.CLIMATE_OPENEO_CREATE, { @@ -125,7 +140,9 @@ export const HANDLERS: Map = new Map< displayName: "DHIS2 Data Upload", description: "Uploads data from a file to DHIS2", tags: ["dhis2"], - schemas: {}, + schemas: { + config: dhis2DataUploadConfigSchema, + }, }, ], [ diff --git a/src/services/worker/services/handlers/index.ts b/src/services/worker/services/handlers/index.ts index 8586d77..a612af4 100644 --- a/src/services/worker/services/handlers/index.ts +++ b/src/services/worker/services/handlers/index.ts @@ -1,3 +1,4 @@ +export * from "./openClimateServiceSync"; export * from "./climateOpenEoCreate"; export * from "./climateOpenEoPoll"; export * from "./climateOpenEoDownload"; diff --git a/src/services/worker/services/handlers/openClimateServiceSync/index.ts b/src/services/worker/services/handlers/openClimateServiceSync/index.ts new file mode 100644 index 0000000..aa27bed --- /dev/null +++ b/src/services/worker/services/handlers/openClimateServiceSync/index.ts @@ -0,0 +1,235 @@ +import { + registerHandler, + type StepContext, + type StepHandler, + type TaskHandle, +} from "@/services/worker/types/service.ts"; +import { Handlers } from "@/services/worker/constants/handlers.ts"; +import { + getClimateJob, + planClimateDatasetSync, + startClimateDatasetSync, + TERMINAL_CLIMATE_JOB_STATUSES, + WORKING_SYNC_ACTIONS, + type ClimateJobStatus, + type ClimateSyncAction, +} from "@/shared/clients/climateSync.ts"; +import { + openClimateServiceSyncConfigSchema, + type OpenClimateServiceSyncConfig, +} from "./schemas/config.ts"; + +export type DatasetSyncOutcome = { + datasetId: string; + outcome: "up-to-date" | "synced" | "failed"; + action?: ClimateSyncAction; + message: string; + /** Last period the dataset covered before this step ran. */ + coveredUntil?: string | null; + /** Period the sync brought the dataset up to, when it downloaded anything. */ + syncedUntil?: string | null; + jobId?: string; +}; + +export type OpenClimateServiceSyncOutput = { + datasets: DatasetSyncOutcome[]; + synced: number; + upToDate: number; + failed: number; +}; + +/** + * Handler: open-climate-service-sync + * + * Brings Open Climate Service datasets up to date before a pipeline reads them, replacing the + * manual monthly sync. Each dataset is its own task and runs alongside the others: + * + * 1. Ask the planner what a sync would do. Nothing to download → done, no job started. + * 2. Otherwise queue the sync and poll its job until it finishes. + * + * A failed dataset fails the step unless `onFailure` is "continue", in which case later steps + * run on the data already there. + */ +export const openClimateServiceSync: StepHandler = { + async execute(ctx) { + const parsed = openClimateServiceSyncConfigSchema.safeParse(ctx.handlerConfig); + if (!parsed.success) { + throw new Error(`Invalid open-climate-service-sync handlerConfig: ${parsed.error.message}`); + } + const config = parsed.data; + + await ctx.log("INFO", "Syncing climate datasets", { + datasetIds: config.datasetIds, + end: config.end ?? "latest", + onFailure: config.onFailure, + }); + + const settled = await Promise.allSettled( + config.datasetIds.map((datasetId) => syncDataset(ctx, datasetId, config)) + ); + + const datasets = settled.map( + (result, index): DatasetSyncOutcome => + result.status === "fulfilled" + ? result.value + : { + datasetId: config.datasetIds[index]!, + outcome: "failed", + message: + result.reason instanceof Error ? result.reason.message : String(result.reason), + } + ); + + const output: OpenClimateServiceSyncOutput = { + datasets, + synced: datasets.filter((d) => d.outcome === "synced").length, + upToDate: datasets.filter((d) => d.outcome === "up-to-date").length, + failed: datasets.filter((d) => d.outcome === "failed").length, + }; + + if (output.failed > 0) { + const failures = datasets + .filter((d) => d.outcome === "failed") + .map((d) => `${d.datasetId}: ${d.message}`) + .join("; "); + if (config.onFailure === "fail") { + throw new Error( + `${output.failed} of ${datasets.length} climate dataset(s) failed to sync — ${failures}` + ); + } + await ctx.log( + "WARN", + "Some climate datasets failed to sync; continuing with the data already available", + { failures } + ); + } + + await ctx.log("INFO", "Open Climate Service sync finished", { + synced: output.synced, + upToDate: output.upToDate, + failed: output.failed, + }); + return output; + }, +}; + +async function syncDataset( + ctx: StepContext, + datasetId: string, + config: OpenClimateServiceSyncConfig +): Promise { + const task = await ctx.tasks.startTask(`sync:${datasetId}`, { + datasetId, + end: config.end ?? null, + }); + + try { + const plan = await planClimateDatasetSync(datasetId, config.end); + await task.log("INFO", "Sync plan", { + action: plan.action, + reason: plan.reason, + message: plan.message, + currentEnd: plan.current_end ?? null, + targetEnd: plan.target_end ?? null, + }); + + if (!WORKING_SYNC_ACTIONS.has(plan.action)) { + const outcome: DatasetSyncOutcome = { + datasetId, + outcome: "up-to-date", + action: plan.action, + message: plan.message, + coveredUntil: plan.current_end ?? null, + }; + if (plan.action === "not_syncable") { + await task.log("WARN", "Dataset cannot be synced; using it as it is", { + reason: plan.reason, + }); + } + await task.succeed(outcome); + return outcome; + } + + const started = await startClimateDatasetSync(datasetId, config.end); + if (started.kind === "completed") { + const outcome: DatasetSyncOutcome = { + datasetId, + outcome: "synced", + action: plan.action, + message: started.message ?? plan.message, + coveredUntil: plan.current_end ?? null, + syncedUntil: plan.target_end ?? null, + }; + await task.succeed(outcome); + return outcome; + } + + await task.log("INFO", "Sync job queued", { jobId: started.jobId }); + await waitForJob(task, datasetId, started.jobId, config.polling); + + const outcome: DatasetSyncOutcome = { + datasetId, + outcome: "synced", + action: plan.action, + message: plan.message, + coveredUntil: plan.current_end ?? null, + syncedUntil: plan.target_end ?? null, + jobId: started.jobId, + }; + await task.succeed(outcome); + return outcome; + } catch (err) { + const error = err instanceof Error ? err : new Error(String(err)); + await task.fail(error); + throw error; + } +} + +async function waitForJob( + task: TaskHandle, + datasetId: string, + jobId: string, + polling: OpenClimateServiceSyncConfig["polling"] +): Promise { + const { pollIntervalMs, maxAttempts } = polling; + let lastStatus: ClimateJobStatus | "unknown" = "unknown"; + + for (let attempt = 1; attempt <= maxAttempts; attempt++) { + await sleep(pollIntervalMs); + + let job; + try { + job = await getClimateJob(jobId); + } catch (err) { + await task.log("WARN", "Could not fetch sync job status, will retry", { + jobId, + attempt, + error: String(err), + }); + continue; + } + + if (job.status !== lastStatus) { + await task.log("INFO", "Sync job status update", { jobId, status: job.status, attempt }); + lastStatus = job.status; + } + + if (TERMINAL_CLIMATE_JOB_STATUSES.has(job.status)) { + if (job.status === "successful") return; + const detail = job.error?.message ? `: ${job.error.message}` : ""; + throw new Error(`Sync job ${jobId} for "${datasetId}" ended as ${job.status}${detail}`); + } + } + + throw new Error( + `Sync job ${jobId} for "${datasetId}" did not finish within ` + + `${(maxAttempts * pollIntervalMs) / 60_000} minutes (last status "${lastStatus}"). ` + + `It keeps running in the Open Climate Service; retry this step once it has finished.` + ); +} + +registerHandler(Handlers.OPEN_CLIMATE_SERVICE_SYNC, openClimateServiceSync); + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} diff --git a/src/services/worker/services/handlers/openClimateServiceSync/schemas/config.ts b/src/services/worker/services/handlers/openClimateServiceSync/schemas/config.ts new file mode 100644 index 0000000..95d9dc6 --- /dev/null +++ b/src/services/worker/services/handlers/openClimateServiceSync/schemas/config.ts @@ -0,0 +1,24 @@ +import { z } from "zod"; + +export const openClimateServiceSyncConfigSchema = z.object({ + datasetIds: z + .array(z.string().min(1)) + .min(1) + .refine((ids) => new Set(ids).size === ids.length, { + message: "Each dataset may only be listed once", + }), + /** + * Last period to sync to. Omitted, each dataset syncs to the latest data its source has + * published, which is what a scheduled run wants. + */ + end: z.iso.date({ message: "Use a date in YYYY-MM-DD format" }).optional(), + onFailure: z.enum(["fail", "continue"]).default("fail"), + polling: z + .object({ + pollIntervalMs: z.number().int().min(1_000).default(10_000), + maxAttempts: z.number().int().min(1).default(150), + }) + .default({ pollIntervalMs: 10_000, maxAttempts: 150 }), +}); + +export type OpenClimateServiceSyncConfig = z.infer; diff --git a/src/shared/clients/climateSync.ts b/src/shared/clients/climateSync.ts new file mode 100644 index 0000000..ea7d6c2 --- /dev/null +++ b/src/shared/clients/climateSync.ts @@ -0,0 +1,161 @@ +import axios from "axios"; +import { z } from "zod"; +import { climateApiClient } from "@/shared/clients/climateApi.ts"; +import { + ASYNC_PREFER_HEADER, + extractJobIdFromLocation, +} from "@/services/api/utils/climate/async.ts"; + +export const climateSyncActionSchema = z.enum(["rematerialize", "append", "no_op", "not_syncable"]); + +export type ClimateSyncAction = z.infer; + +/** Planner actions that download data; the others leave the dataset as it is. */ +export const WORKING_SYNC_ACTIONS = new Set(["append", "rematerialize"]); + +export const climateSyncPlanSchema = z + .object({ + action: climateSyncActionSchema, + reason: z.string(), + message: z.string(), + current_start: z.string().nullable().optional(), + current_end: z.string().nullable().optional(), + target_end: z.string().nullable().optional(), + delta_start: z.string().nullable().optional(), + delta_end: z.string().nullable().optional(), + }) + .passthrough(); + +export type ClimateSyncPlan = z.infer; + +export const climateJobStatusSchema = z.enum([ + "accepted", + "running", + "retrying", + "successful", + "failed", + "cancelled", +]); + +export type ClimateJobStatus = z.infer; + +export const TERMINAL_CLIMATE_JOB_STATUSES = new Set([ + "successful", + "failed", + "cancelled", +]); + +export const climateJobRecordSchema = z + .object({ + jobID: z.string(), + status: climateJobStatusSchema, + error: z.object({ type: z.string(), message: z.string() }).nullable().optional(), + }) + .passthrough(); + +export type ClimateJobRecord = z.infer; + +/** A sync either queued as a background job, or finished inline (e.g. nothing to download). */ +export type StartedClimateSync = + | { kind: "queued"; jobId: string } + | { kind: "completed"; status: string; message: string | null }; + +const syncCompletedResponseSchema = z + .object({ status: z.string(), message: z.string().nullable().optional() }) + .passthrough(); + +/** + * Rewrites a climate API failure as an error naming the dataset and the API's own detail, + * so it reads well in the step log instead of as a bare "Request failed with status code". + */ +function describeClimateError(action: string, datasetId: string, err: unknown): Error { + if (axios.isAxiosError(err) && err.response) { + const status = err.response.status; + const data: unknown = err.response.data; + const detail = + typeof data === "object" && data !== null && "detail" in data + ? JSON.stringify((data as { detail: unknown }).detail) + : JSON.stringify(data); + const hint = + status === 404 + ? " — the dataset no longer exists in the Open Climate Service" + : status === 409 + ? " — a sync for this dataset is probably already running; retry once it finishes" + : ""; + return new Error( + `Could not ${action} climate dataset "${datasetId}" (${status}): ${detail}${hint}`, + { cause: err } + ); + } + const message = err instanceof Error ? err.message : String(err); + return new Error(`Could not ${action} climate dataset "${datasetId}": ${message}`, { + cause: err, + }); +} + +/** GET /sync/{dataset_id}/plan — what a sync would do, without downloading anything. */ +export async function planClimateDatasetSync( + datasetId: string, + end?: string +): Promise { + let data: unknown; + try { + const response = await climateApiClient.get(`/sync/${encodeURIComponent(datasetId)}/plan`, { + params: end !== undefined ? { end } : undefined, + }); + data = response.data; + } catch (err) { + throw describeClimateError("plan a sync for", datasetId, err); + } + const parsed = climateSyncPlanSchema.safeParse(data); + if (!parsed.success) { + throw new Error( + `Invalid sync plan for climate dataset "${datasetId}": ${parsed.error.message}` + ); + } + return parsed.data; +} + +/** POST /sync/{dataset_id} with `Prefer: respond-async`, publishing the synced version. */ +export async function startClimateDatasetSync( + datasetId: string, + end?: string +): Promise { + try { + const response = await climateApiClient.post( + `/sync/${encodeURIComponent(datasetId)}`, + { end: end ?? null, publish: true }, + { headers: { Prefer: ASYNC_PREFER_HEADER } } + ); + if (response.status === 202) { + const headers = response.headers as Record; + const location = typeof headers.location === "string" ? headers.location : undefined; + const jobId = extractJobIdFromLocation(location); + if (!jobId) { + throw new Error( + `Sync for climate dataset "${datasetId}" was accepted without a job location` + ); + } + return { kind: "queued", jobId }; + } + const parsed = syncCompletedResponseSchema.safeParse(response.data); + return { + kind: "completed", + status: parsed.success ? parsed.data.status : "completed", + message: parsed.success ? (parsed.data.message ?? null) : null, + }; + } catch (err) { + if (err instanceof Error && !axios.isAxiosError(err)) throw err; + throw describeClimateError("start a sync for", datasetId, err); + } +} + +/** GET /ingestions/jobs/{job_id} */ +export async function getClimateJob(jobId: string): Promise { + const response = await climateApiClient.get(`/ingestions/jobs/${encodeURIComponent(jobId)}`); + const parsed = climateJobRecordSchema.safeParse(response.data); + if (!parsed.success) { + throw new Error(`Invalid climate job record for ${jobId}: ${parsed.error.message}`); + } + return parsed.data; +} diff --git a/src/tests/unit/handlers-catalog.test.ts b/src/tests/unit/handlers-catalog.test.ts index b300390..27bafe1 100644 --- a/src/tests/unit/handlers-catalog.test.ts +++ b/src/tests/unit/handlers-catalog.test.ts @@ -12,16 +12,18 @@ import { predictionTriggerConfigSchema } from "@/services/worker/services/handle import { thresholdGenerationConfigSchema } from "@/services/worker/services/handlers/thresholdGeneration/schemas/config.ts"; import { alertGenerationConfigSchema } from "@/services/worker/services/handlers/alertGeneration/schemas/config.ts"; import { predictionDataDownloadConfigSchema } from "@/services/worker/services/handlers/predictionDataDownload/schemas/config.ts"; +import { dhis2DataUploadConfigSchema } from "@/services/worker/services/handlers/dhis2DataUpload/schemas/config.ts"; import { dhis2AnalyticsRunConfigSchema } from "@/services/worker/services/handlers/dhis2AnalyticsRun/schemas/config.ts"; describe("handler catalog", () => { it("lists all registered handlers with display metadata", () => { const handlers = listHandlerDescriptors(); - expect(handlers).toHaveLength(11); + expect(handlers).toHaveLength(12); const keys = handlers.map((h) => h.key).sort(); expect(keys).toEqual( [ "alert-generation", + "open-climate-service-sync", "climate-openeo-create", "climate-openeo-download", "climate-openeo-poll", @@ -115,6 +117,11 @@ describe("handler config JSON Schema round-trip", () => { valueDataElementIds: ["DE_VALUE"], }, }, + { + name: "dhis2-data-upload", + schema: dhis2DataUploadConfigSchema, + valid: { importStrategy: "CREATE" }, + }, { name: "dhis2-analytics-run", schema: dhis2AnalyticsRunConfigSchema, diff --git a/src/tests/unit/handlers/openClimateServiceSync.test.ts b/src/tests/unit/handlers/openClimateServiceSync.test.ts new file mode 100644 index 0000000..2c16832 --- /dev/null +++ b/src/tests/unit/handlers/openClimateServiceSync.test.ts @@ -0,0 +1,237 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import { AxiosError, AxiosHeaders } from "axios"; +import { buildMockContext } from "@/tests/utils/helpers.ts"; + +vi.mock("@/shared/clients/climateApi.ts", () => ({ + climateApiClient: { get: vi.fn(), post: vi.fn() }, +})); + +import { climateApiClient } from "@/shared/clients/climateApi.ts"; +import { + openClimateServiceSync, + type OpenClimateServiceSyncOutput, +} from "@/services/worker/services/handlers/openClimateServiceSync/index.ts"; +import { openClimateServiceSyncConfigSchema } from "@/services/worker/services/handlers/openClimateServiceSync/schemas/config.ts"; + +const mockGet = vi.mocked(climateApiClient.get); +const mockPost = vi.mocked(climateApiClient.post); + +const polling = { pollIntervalMs: 1_000, maxAttempts: 5 }; + +function plan(action: string, extra: Record = {}) { + return { + data: { + source_dataset_id: "tpl", + sync_kind: "temporal", + action, + reason: `reason_${action}`, + message: `planned ${action}`, + target_end_source: "default_today", + current_end: "2026-07", + target_end: "2026-08", + ...extra, + }, + }; +} + +function accepted(jobId: string) { + return { + status: 202, + headers: { location: `/ingestions/jobs/${jobId}` }, + data: { status: "accepted" }, + }; +} + +function job(jobID: string, status: string, error?: { type: string; message: string }) { + return { data: { jobID, processID: "sync", status, error: error ?? null } }; +} + +/** Routes GETs by URL: `/sync/{id}/plan` → plans[id], `/ingestions/jobs/{id}` → next of jobs[id]. */ +function routeGets(plans: Record, jobs: Record = {}): void { + mockGet.mockImplementation(async (url: string) => { + const planMatch = url.match(/^\/sync\/([^/]+)\/plan$/); + if (planMatch) { + const p = plans[decodeURIComponent(planMatch[1]!)]; + if (p instanceof Error) throw p; + return p; + } + const jobMatch = url.match(/^\/ingestions\/jobs\/([^/]+)$/); + if (jobMatch) { + const queue = jobs[jobMatch[1]!]!; + const next = queue.length > 1 ? queue.shift() : queue[0]; + if (next instanceof Error) throw next; + return next; + } + throw new Error(`unexpected GET ${url}`); + }); +} + +async function run(handlerConfig: Record) { + const ctx = buildMockContext({ handlerConfig }); + const promise = openClimateServiceSync.execute(ctx); + // Keep the rejection observed while timers advance; the test awaits it afterwards. + promise.catch(() => undefined); + await vi.advanceTimersByTimeAsync(60_000); + return { ctx, promise }; +} + +describe("open-climate-service-sync handler", () => { + beforeEach(() => { + vi.clearAllMocks(); + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("skips datasets the planner reports as up to date without starting a sync", async () => { + routeGets({ temp: plan("no_op", { current_end: "2026-08" }) }); + + const { promise } = await run({ datasetIds: ["temp"], polling }); + const result = (await promise) as OpenClimateServiceSyncOutput; + + expect(mockPost).not.toHaveBeenCalled(); + expect(result).toMatchObject({ synced: 0, upToDate: 1, failed: 0 }); + expect(result.datasets[0]).toMatchObject({ + datasetId: "temp", + outcome: "up-to-date", + coveredUntil: "2026-08", + }); + }); + + it("queues a sync and polls its job until it succeeds", async () => { + routeGets( + { temp: plan("append") }, + { job_1: [job("job_1", "running"), job("job_1", "successful")] } + ); + mockPost.mockResolvedValue(accepted("job_1")); + + const { promise } = await run({ datasetIds: ["temp"], end: "2026-08-31", polling }); + const result = (await promise) as OpenClimateServiceSyncOutput; + + expect(mockPost).toHaveBeenCalledWith( + "/sync/temp", + { end: "2026-08-31", publish: true }, + { headers: { Prefer: "respond-async" } } + ); + expect(result.datasets[0]).toMatchObject({ + outcome: "synced", + action: "append", + coveredUntil: "2026-07", + syncedUntil: "2026-08", + jobId: "job_1", + }); + }); + + it("syncs each dataset as its own task", async () => { + routeGets( + { temp: plan("no_op"), precip: plan("append") }, + { job_p: [job("job_p", "successful")] } + ); + mockPost.mockResolvedValue(accepted("job_p")); + + const { ctx, promise } = await run({ datasetIds: ["temp", "precip"], polling }); + const result = (await promise) as OpenClimateServiceSyncOutput; + + expect(vi.mocked(ctx.tasks.startTask).mock.calls.map(([name]) => name)).toEqual([ + "sync:temp", + "sync:precip", + ]); + expect(result).toMatchObject({ synced: 1, upToDate: 1, failed: 0 }); + expect(mockPost).toHaveBeenCalledTimes(1); + }); + + it("fails the step when a sync job fails, naming the dataset and the job's error", async () => { + routeGets( + { temp: plan("no_op"), precip: plan("append") }, + { job_p: [job("job_p", "failed", { type: "DownloadError", message: "CHIRPS timed out" })] } + ); + mockPost.mockResolvedValue(accepted("job_p")); + + const { promise } = await run({ datasetIds: ["temp", "precip"], polling }); + + await expect(promise).rejects.toThrow( + /1 of 2 climate dataset\(s\) failed to sync — precip: Sync job job_p for "precip" ended as failed: CHIRPS timed out/ + ); + }); + + it("continues past a failed dataset when onFailure is continue", async () => { + routeGets({ temp: plan("no_op"), precip: new Error("socket hang up") }); + + const { ctx, promise } = await run({ + datasetIds: ["temp", "precip"], + onFailure: "continue", + polling, + }); + const result = (await promise) as OpenClimateServiceSyncOutput; + + expect(result).toMatchObject({ upToDate: 1, failed: 1 }); + expect(result.datasets[1]).toMatchObject({ datasetId: "precip", outcome: "failed" }); + expect(ctx.log).toHaveBeenCalledWith( + "WARN", + expect.stringContaining("continuing with the data already available"), + expect.anything() + ); + }); + + it("explains a conflict as a sync that is already running", async () => { + routeGets({ temp: plan("append") }); + mockPost.mockRejectedValue( + new AxiosError("conflict", "ERR_BAD_REQUEST", undefined, undefined, { + status: 409, + statusText: "Conflict", + data: { detail: "sync in progress" }, + headers: {}, + config: { headers: new AxiosHeaders() }, + }) + ); + + const { promise } = await run({ datasetIds: ["temp"], polling }); + + await expect(promise).rejects.toThrow(/already running/); + }); + + it("keeps polling through transient job-status errors", async () => { + routeGets({ temp: plan("append") }, { job_1: [new Error("502"), job("job_1", "successful")] }); + mockPost.mockResolvedValue(accepted("job_1")); + + const { promise } = await run({ datasetIds: ["temp"], polling }); + const result = (await promise) as OpenClimateServiceSyncOutput; + + expect(result.synced).toBe(1); + }); + + it("times out with a message saying the job keeps running", async () => { + routeGets({ temp: plan("append") }, { job_1: [job("job_1", "running")] }); + mockPost.mockResolvedValue(accepted("job_1")); + + const { promise } = await run({ datasetIds: ["temp"], polling }); + + await expect(promise).rejects.toThrow(/keeps running in the Open Climate Service/); + }); +}); + +describe("open-climate-service-sync config", () => { + it("defaults to failing on error and to polling under the consumer timeout", () => { + const config = openClimateServiceSyncConfigSchema.parse({ datasetIds: ["temp"] }); + expect(config.onFailure).toBe("fail"); + expect(config.polling.pollIntervalMs * config.polling.maxAttempts).toBeLessThan(30 * 60_000); + }); + + it("rejects an empty or duplicated dataset list", () => { + expect(openClimateServiceSyncConfigSchema.safeParse({ datasetIds: [] }).success).toBe(false); + expect( + openClimateServiceSyncConfigSchema.safeParse({ datasetIds: ["temp", "temp"] }).success + ).toBe(false); + }); + + it("accepts only a real YYYY-MM-DD end date", () => { + const parse = (end: string) => + openClimateServiceSyncConfigSchema.safeParse({ datasetIds: ["temp"], end }).success; + expect(parse("2026-08-31")).toBe(true); + expect(parse("2026-08")).toBe(false); + expect(parse("2026-13-01")).toBe(false); + expect(parse("aug 2026")).toBe(false); + }); +}); From 66164c19ceff910919f0943d5a652185c5bcbae9 Mon Sep 17 00:00:00 2001 From: Allen7702 Date: Sat, 26 Sep 2026 12:52:28 +0300 Subject: [PATCH 2/5] feat: add import strategy configuration for DHIS2 data upload handler --- .../handlers/dhis2DataUpload/index.ts | 44 ++++++---- .../dhis2DataUpload/schemas/config.ts | 17 ++++ src/services/worker/utils/dhis2.ts | 41 ++++++--- src/tests/unit/postDataValueSet.test.ts | 88 +++++++++++++++++++ src/tests/unit/predictionDataDownload.test.ts | 76 ++++++++++++++++ 5 files changed, 238 insertions(+), 28 deletions(-) create mode 100644 src/services/worker/services/handlers/dhis2DataUpload/schemas/config.ts create mode 100644 src/tests/unit/postDataValueSet.test.ts diff --git a/src/services/worker/services/handlers/dhis2DataUpload/index.ts b/src/services/worker/services/handlers/dhis2DataUpload/index.ts index 460245d..45ba7b6 100644 --- a/src/services/worker/services/handlers/dhis2DataUpload/index.ts +++ b/src/services/worker/services/handlers/dhis2DataUpload/index.ts @@ -4,7 +4,7 @@ import { env } from "@/shared/utils/env.ts"; import path from "node:path"; import type { DataValueSet, ImportSummary } from "@/services/worker/types/data.ts"; import { postDataValueSet } from "@/services/worker/utils/dhis2.ts"; -import { AxiosError } from "axios"; +import { dhis2DataUploadConfigSchema } from "@/services/worker/services/handlers/dhis2DataUpload/schemas/config.ts"; /** * Handler: dhis2-data-upload @@ -15,11 +15,19 @@ import { AxiosError } from "axios"; * Requires ctx.input.filename — basename under OUTPUTS_DIR from the previous step * (e.g. threshold-generation or prediction-data-download). * + * Config: { importStrategy } — CREATE_AND_UPDATE (default), CREATE or UPDATE. + * * Output: * { status: string; imported: number; updated: number; ignored: number; deleted: number } */ export const dhis2DataUpload: StepHandler = { async execute(ctx) { + const parsed = dhis2DataUploadConfigSchema.safeParse(ctx.handlerConfig ?? {}); + if (!parsed.success) { + throw new Error(`Invalid dhis2-data-upload handlerConfig: ${parsed.error.message}`); + } + const { importStrategy } = parsed.data; + const input = ctx.input as { filename?: string } | undefined; const filename = input?.filename; @@ -35,6 +43,7 @@ export const dhis2DataUpload: StepHandler = { const task = await ctx.tasks.startTask("upload-data-value-set", { filePath, + importStrategy, }); let payload: DataValueSet; @@ -58,36 +67,35 @@ export const dhis2DataUpload: StepHandler = { await ctx.log("INFO", "Uploading data values to DHIS2", { filePath, count: payload.dataValues.length, + importStrategy, }); - let summary; + let summary: ImportSummary; try { - summary = await postDataValueSet({ payload, ctx }); + summary = await postDataValueSet({ payload, ctx, importStrategy }); } catch (err) { - if (err instanceof AxiosError) { - if (err.response?.status === 409) { - summary = err.response.data as ImportSummary; - await task.fail(err); - } else { - await task.fail(err); - throw err; - } - } else { - const error = err instanceof Error ? err : new Error(String(err)); - await task.fail(error); - throw error; - } + const error = err instanceof Error ? err : new Error(String(err)); + await task.fail(error); + throw error; } + const conflicts = summary.conflicts?.map((c) => `${c.object}: ${c.value}`) ?? []; + if (summary.status === "ERROR") { - const conflicts = summary.conflicts?.map((c) => `${c.object}: ${c.value}`).join("; "); const error = new Error( - `DHIS2 data value import failed with status ERROR${conflicts ? `: ${conflicts}` : ""}` + `DHIS2 data value import failed with status ERROR${conflicts.length ? `: ${conflicts.join("; ")}` : ""}` ); await task.fail(error); throw error; } + if (conflicts.length > 0) { + await ctx.log("WARN", "DHIS2 data value import completed with conflicts", { + status: summary.status, + conflicts, + }); + } + const counts = summary.importCount ?? { imported: 0, updated: 0, diff --git a/src/services/worker/services/handlers/dhis2DataUpload/schemas/config.ts b/src/services/worker/services/handlers/dhis2DataUpload/schemas/config.ts new file mode 100644 index 0000000..2e354b7 --- /dev/null +++ b/src/services/worker/services/handlers/dhis2DataUpload/schemas/config.ts @@ -0,0 +1,17 @@ +import { z } from "zod"; + +/** + * `importStrategy` query parameter for DHIS2 `POST /dataValueSets`. + * - CREATE_AND_UPDATE (Merge): import new values and update existing + * - CREATE (Append): import new values only + * - UPDATE (Update): only update existing values, ignore new values + */ +export const dataValueImportStrategySchema = z.enum(["CREATE_AND_UPDATE", "CREATE", "UPDATE"]); + +export type DataValueImportStrategy = z.infer; + +export const dhis2DataUploadConfigSchema = z.object({ + importStrategy: dataValueImportStrategySchema.default("CREATE_AND_UPDATE"), +}); + +export type Dhis2DataUploadConfig = z.infer; diff --git a/src/services/worker/utils/dhis2.ts b/src/services/worker/utils/dhis2.ts index d2e279f..280b61c 100644 --- a/src/services/worker/utils/dhis2.ts +++ b/src/services/worker/utils/dhis2.ts @@ -5,6 +5,7 @@ import type { StepContext } from "@/services/worker/types/service.ts"; import { chunk, groupBy } from "lodash-es"; import { PeriodUtility } from "@hisptz/dhis2-utils"; import { AxiosError } from "axios"; +import type { DataValueImportStrategy } from "@/services/worker/services/handlers/dhis2DataUpload/schemas/config.ts"; export type TrackerImportStrategy = "CREATE" | "UPDATE" | "CREATE_AND_UPDATE" | "DELETE"; @@ -46,33 +47,53 @@ export async function postTrackerEvents({ } } +/** + * DHIS2 2.38+ wraps the import summary in a WebMessage envelope + * (`{ httpStatus, status, message, response: ImportSummary }`); older versions + * return the ImportSummary directly. Accept both shapes. + */ +export function unwrapImportSummary(data: unknown): ImportSummary { + const inner = (data as { response?: unknown } | null)?.response; + if ( + inner && + typeof inner === "object" && + ((inner as { responseType?: string }).responseType === "ImportSummary" || + "importCount" in inner) + ) { + return inner as ImportSummary; + } + return data as ImportSummary; +} + /** * POST /dataValueSets * Imports a set of aggregate data values into DHIS2. + * + * A 409 from DHIS2 still carries an import summary (status ERROR plus conflicts), + * so it is returned rather than thrown; the caller decides how to fail. */ export async function postDataValueSet({ payload, ctx, + importStrategy = "CREATE_AND_UPDATE", }: { payload: DataValueSet; ctx: StepContext; + importStrategy?: DataValueImportStrategy; }): Promise { try { - const response = await dhis2RestClient.post("/dataValueSets", payload); - return response.data; + const response = await dhis2RestClient.post("/dataValueSets", payload, { + params: { importStrategy }, + }); + return unwrapImportSummary(response.data); } catch (error) { + if (error instanceof AxiosError && error.response?.status === 409 && error.response.data) { + return unwrapImportSummary(error.response.data); + } await ctx.log( "ERROR", `Failed to upload data values: ${error instanceof Error ? error.message : "Unknown Error"}` ); - if (error instanceof AxiosError) { - if (error.response?.status === 409) { - await ctx.log( - "INFO", - `Conflict error occurred during data upload. ${JSON.stringify(error.response?.data)}` - ); - } - } throw error; } } diff --git a/src/tests/unit/postDataValueSet.test.ts b/src/tests/unit/postDataValueSet.test.ts new file mode 100644 index 0000000..de0a282 --- /dev/null +++ b/src/tests/unit/postDataValueSet.test.ts @@ -0,0 +1,88 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { AxiosError, AxiosHeaders } from "axios"; +import { buildMockContext } from "@/tests/utils/helpers.ts"; + +vi.mock("@/shared/clients/dhis.ts", () => ({ + dhis2RestClient: { post: vi.fn() }, +})); + +import { dhis2RestClient } from "@/shared/clients/dhis.ts"; +import { postDataValueSet, unwrapImportSummary } from "@/services/worker/utils/dhis2.ts"; + +const mockPost = vi.mocked(dhis2RestClient.post); + +const importSummary = { + responseType: "ImportSummary", + status: "SUCCESS", + importCount: { imported: 12, updated: 3, ignored: 0, deleted: 0 }, + conflicts: [], +}; + +const webMessage = (response: unknown, status = "OK") => ({ + httpStatus: status === "OK" ? "OK" : "Conflict", + httpStatusCode: status === "OK" ? 200 : 409, + status, + message: "Import finished", + response, +}); + +function conflictError(data: unknown): AxiosError { + return new AxiosError( + "Request failed with status code 409", + "ERR_BAD_REQUEST", + undefined, + undefined, + { + status: 409, + statusText: "Conflict", + headers: {}, + config: { headers: new AxiosHeaders() }, + data, + } + ); +} + +const payload = { + dataValues: [{ dataElement: "de", period: "202501", orgUnit: "ou", value: "1" }], +}; + +describe("unwrapImportSummary", () => { + it("unwraps the DHIS2 2.38+ WebMessage envelope", () => { + expect(unwrapImportSummary(webMessage(importSummary))).toEqual(importSummary); + }); + + it("returns a legacy bare ImportSummary unchanged", () => { + expect(unwrapImportSummary(importSummary)).toEqual(importSummary); + }); +}); + +describe("postDataValueSet", () => { + beforeEach(() => vi.clearAllMocks()); + + it("returns the inner import summary with real counts", async () => { + mockPost.mockResolvedValue({ data: webMessage(importSummary) }); + const summary = await postDataValueSet({ payload, ctx: buildMockContext() }); + expect(summary.status).toBe("SUCCESS"); + expect(summary.importCount).toEqual({ imported: 12, updated: 3, ignored: 0, deleted: 0 }); + }); + + it("returns the summary from a 409 instead of throwing", async () => { + const errorSummary = { + responseType: "ImportSummary", + status: "ERROR", + importCount: { imported: 0, updated: 0, ignored: 1, deleted: 0 }, + conflicts: [{ object: "de", value: "Data element not found" }], + }; + mockPost.mockRejectedValue(conflictError(webMessage(errorSummary, "ERROR"))); + const summary = await postDataValueSet({ payload, ctx: buildMockContext() }); + expect(summary.status).toBe("ERROR"); + expect(summary.conflicts).toEqual(errorSummary.conflicts); + }); + + it("rethrows non-409 errors", async () => { + mockPost.mockRejectedValue(new Error("network down")); + await expect(postDataValueSet({ payload, ctx: buildMockContext() })).rejects.toThrow( + "network down" + ); + }); +}); diff --git a/src/tests/unit/predictionDataDownload.test.ts b/src/tests/unit/predictionDataDownload.test.ts index 1784e49..a44969a 100644 --- a/src/tests/unit/predictionDataDownload.test.ts +++ b/src/tests/unit/predictionDataDownload.test.ts @@ -180,6 +180,47 @@ describe("dhis2DataUpload filename resolution", () => { ); }); + it("defaults importStrategy to CREATE_AND_UPDATE when config omits it", async () => { + mockFileJson.mockResolvedValue({ + dataValues: [{ dataElement: "de", period: "202501", orgUnit: "ou", value: "1" }], + }); + + const ctx = buildMockContext({ + input: { filename: "from-prev-step.json" }, + handlerConfig: {}, + }); + await dhis2DataUpload.execute(ctx); + + expect(mockPostDataValueSet).toHaveBeenCalledWith( + expect.objectContaining({ importStrategy: "CREATE_AND_UPDATE" }) + ); + }); + + it("passes the configured importStrategy to DHIS2", async () => { + mockFileJson.mockResolvedValue({ + dataValues: [{ dataElement: "de", period: "202501", orgUnit: "ou", value: "1" }], + }); + + const ctx = buildMockContext({ + input: { filename: "from-prev-step.json" }, + handlerConfig: { importStrategy: "UPDATE" }, + }); + await dhis2DataUpload.execute(ctx); + + expect(mockPostDataValueSet).toHaveBeenCalledWith( + expect.objectContaining({ importStrategy: "UPDATE" }) + ); + }); + + it("rejects an unknown importStrategy", async () => { + const ctx = buildMockContext({ + input: { filename: "from-prev-step.json" }, + handlerConfig: { importStrategy: "DELETE" }, + }); + await expect(dhis2DataUpload.execute(ctx)).rejects.toThrow(/handlerConfig/); + expect(mockPostDataValueSet).not.toHaveBeenCalled(); + }); + it("throws when prior step output omits filename", async () => { const ctx = buildMockContext({ input: {}, @@ -188,6 +229,41 @@ describe("dhis2DataUpload filename resolution", () => { await expect(dhis2DataUpload.execute(ctx)).rejects.toThrow(/filename/); }); + it("fails with DHIS2 conflicts when the import status is ERROR", async () => { + mockFileJson.mockResolvedValue({ + dataValues: [{ dataElement: "de", period: "202501", orgUnit: "ou", value: "1" }], + }); + mockPostDataValueSet.mockResolvedValue({ + status: "ERROR", + importCount: { imported: 0, updated: 0, ignored: 1, deleted: 0 }, + conflicts: [{ object: "de", value: "Data element not found" }], + }); + + const ctx = buildMockContext({ input: { filename: "f.json" }, handlerConfig: {} }); + await expect(dhis2DataUpload.execute(ctx)).rejects.toThrow(/de: Data element not found/); + }); + + it("returns real counts and logs conflicts on WARNING", async () => { + mockFileJson.mockResolvedValue({ + dataValues: [{ dataElement: "de", period: "202501", orgUnit: "ou", value: "1" }], + }); + mockPostDataValueSet.mockResolvedValue({ + status: "WARNING", + importCount: { imported: 5, updated: 2, ignored: 1, deleted: 0 }, + conflicts: [{ object: "ou", value: "Org unit not in hierarchy" }], + }); + + const ctx = buildMockContext({ input: { filename: "f.json" }, handlerConfig: {} }); + const result = await dhis2DataUpload.execute(ctx); + + expect(result).toEqual({ status: "WARNING", imported: 5, updated: 2, ignored: 1, deleted: 0 }); + expect(ctx.log).toHaveBeenCalledWith( + "WARN", + expect.stringMatching(/conflicts/), + expect.objectContaining({ conflicts: ["ou: Org unit not in hierarchy"] }) + ); + }); + it("ignores handlerConfig.filename if present", async () => { const jsonPayload = { dataValues: [{ dataElement: "de", period: "202501", orgUnit: "ou", value: "1" }], From a82ed3d5b45e6fb9c89f849432438f6d9f9e5958 Mon Sep 17 00:00:00 2001 From: Allen7702 Date: Sat, 26 Sep 2026 12:59:09 +0300 Subject: [PATCH 3/5] fix: add structured error handling with detailed context for DHIS2 data uploads --- .../migration.sql | 4 + prisma/schemas/run.prisma | 2 + src/services/api/utils/monitoring/queries.ts | 2 +- .../handlers/dhis2DataUpload/index.ts | 15 ++- src/services/worker/services/runners/step.ts | 12 +- src/services/worker/services/workers/step.ts | 7 +- src/services/worker/types/data.ts | 1 + src/shared/utils/error.ts | 108 ++++++++++++++++++ src/tests/unit/errorRecord.test.ts | 68 +++++++++++ src/tests/unit/predictionDataDownload.test.ts | 16 ++- src/tests/utils/helpers.ts | 2 + 11 files changed, 221 insertions(+), 16 deletions(-) create mode 100644 prisma/migrations/20260926120000_add_error_details/migration.sql create mode 100644 src/tests/unit/errorRecord.test.ts diff --git a/prisma/migrations/20260926120000_add_error_details/migration.sql b/prisma/migrations/20260926120000_add_error_details/migration.sql new file mode 100644 index 0000000..b4f2f31 --- /dev/null +++ b/prisma/migrations/20260926120000_add_error_details/migration.sql @@ -0,0 +1,4 @@ +-- Structured failure context kept alongside the short errorMessage headline, so +-- long upstream errors (e.g. hundreds of DHIS2 conflicts) don't live in one string. +ALTER TABLE "step_executions" ADD COLUMN "errorDetails" JSONB; +ALTER TABLE "task_executions" ADD COLUMN "errorDetails" JSONB; diff --git a/prisma/schemas/run.prisma b/prisma/schemas/run.prisma index 7205f74..5726c80 100644 --- a/prisma/schemas/run.prisma +++ b/prisma/schemas/run.prisma @@ -63,6 +63,7 @@ model StepExecution { errorMessage String? errorStack String? + errorDetails Json? // Set once when the coordinator has processed this attempt's result resultProcessedAt DateTime? @@ -140,6 +141,7 @@ model TaskExecution { errorMessage String? errorStack String? + errorDetails Json? startedAt DateTime? finishedAt DateTime? diff --git a/src/services/api/utils/monitoring/queries.ts b/src/services/api/utils/monitoring/queries.ts index 14022c8..8c52621 100644 --- a/src/services/api/utils/monitoring/queries.ts +++ b/src/services/api/utils/monitoring/queries.ts @@ -323,7 +323,7 @@ export class MonitoringQueries { const rows = await this.prisma.$queryRaw<{ error_message: string; occurrence_count: bigint }[]>` SELECT - "errorMessage", + "errorMessage" AS error_message, COUNT(*) AS occurrence_count FROM step_executions WHERE status = 'FAILED' diff --git a/src/services/worker/services/handlers/dhis2DataUpload/index.ts b/src/services/worker/services/handlers/dhis2DataUpload/index.ts index 45ba7b6..94ca1c8 100644 --- a/src/services/worker/services/handlers/dhis2DataUpload/index.ts +++ b/src/services/worker/services/handlers/dhis2DataUpload/index.ts @@ -4,6 +4,7 @@ import { env } from "@/shared/utils/env.ts"; import path from "node:path"; import type { DataValueSet, ImportSummary } from "@/services/worker/types/data.ts"; import { postDataValueSet } from "@/services/worker/utils/dhis2.ts"; +import { groupConflicts, StepError } from "@/shared/utils/error.ts"; import { dhis2DataUploadConfigSchema } from "@/services/worker/services/handlers/dhis2DataUpload/schemas/config.ts"; /** @@ -79,12 +80,15 @@ export const dhis2DataUpload: StepHandler = { throw error; } - const conflicts = summary.conflicts?.map((c) => `${c.object}: ${c.value}`) ?? []; + const conflicts = summary.conflicts ?? []; if (summary.status === "ERROR") { - const error = new Error( - `DHIS2 data value import failed with status ERROR${conflicts.length ? `: ${conflicts.join("; ")}` : ""}` - ); + const error = new StepError("DHIS2 rejected the data value import", { + source: "dhis2", + description: summary.description, + totalConflicts: conflicts.length, + conflicts: groupConflicts(conflicts), + }); await task.fail(error); throw error; } @@ -92,7 +96,8 @@ export const dhis2DataUpload: StepHandler = { if (conflicts.length > 0) { await ctx.log("WARN", "DHIS2 data value import completed with conflicts", { status: summary.status, - conflicts, + totalConflicts: conflicts.length, + conflicts: groupConflicts(conflicts), }); } diff --git a/src/services/worker/services/runners/step.ts b/src/services/worker/services/runners/step.ts index 40f5c12..6090dc5 100644 --- a/src/services/worker/services/runners/step.ts +++ b/src/services/worker/services/runners/step.ts @@ -18,6 +18,7 @@ import type { InputJsonValue } from "@prisma/client/runtime/client"; import { SpanStatusCode } from "@opentelemetry/api"; import { tracer, stepExecutionCounter, stepDurationHistogram } from "@/shared/telemetry/index.ts"; import { logger } from "@/shared/utils"; +import { toErrorRecord } from "@/shared/utils/error.ts"; import { resolveEffectiveHandlerConfig } from "@/services/worker/utils/resolveEffectiveHandlerConfig.ts"; import { parseStepSnapshot, @@ -95,8 +96,7 @@ export function createTaskReporter( where: { id: taskExecution.id }, data: { status: TaskExecutionStatus.FAILED, - errorMessage: error.message, - errorStack: error.stack, + ...errorColumns(error), finishedAt: new Date(), }, }); @@ -106,6 +106,11 @@ export function createTaskReporter( }; } +export function errorColumns(err: Error) { + const { errorMessage, errorStack, errorDetails } = toErrorRecord(err); + return { errorMessage, errorStack, errorDetails: errorDetails as InputJsonValue | undefined }; +} + function createStepLogger( prisma: PrismaClient, execution: PipelineExecution, @@ -356,8 +361,7 @@ export class StepRunner { where: { id: stepExecution.id }, data: { status: StepExecutionStatus.FAILED, - errorMessage: err.message, - errorStack: err.stack, + ...errorColumns(err), finishedAt: new Date(), }, }); diff --git a/src/services/worker/services/workers/step.ts b/src/services/worker/services/workers/step.ts index 435d44b..9ab6783 100644 --- a/src/services/worker/services/workers/step.ts +++ b/src/services/worker/services/workers/step.ts @@ -8,7 +8,7 @@ import { type StepResultMessage, } from "@/services/worker/services/runners/pipeline.ts"; import type { InputJsonValue } from "@prisma/client/runtime/client"; -import { createTaskReporter } from "@/services/worker/services/runners/step.ts"; +import { createTaskReporter, errorColumns } from "@/services/worker/services/runners/step.ts"; import { resolveEffectiveHandlerConfig } from "@/services/worker/utils/resolveEffectiveHandlerConfig.ts"; import { parseStepSnapshot, @@ -184,8 +184,7 @@ export class StepWorker { where: { id: stepExecutionId }, data: { status: StepExecutionStatus.FAILED, - errorMessage: err.message, - errorStack: err.stack, + ...errorColumns(err), finishedAt: new Date(), }, }); @@ -196,7 +195,7 @@ export class StepWorker { executionId, stepExecutionId, succeeded: false, - errorMessage: err.message, + errorMessage: errorColumns(err).errorMessage, } satisfies StepResultMessage, }); logger.error( diff --git a/src/services/worker/types/data.ts b/src/services/worker/types/data.ts index 8236b7a..c30fdda 100644 --- a/src/services/worker/types/data.ts +++ b/src/services/worker/types/data.ts @@ -21,6 +21,7 @@ export interface DataValueSet { } export interface ImportSummary { status: string; + description?: string; importOptions?: Record; importCount?: { imported: number; diff --git a/src/shared/utils/error.ts b/src/shared/utils/error.ts index e69de29..7e4fec5 100644 --- a/src/shared/utils/error.ts +++ b/src/shared/utils/error.ts @@ -0,0 +1,108 @@ +import { AxiosError } from "axios"; + +export const MAX_ERROR_MESSAGE_LENGTH = 300; +const MAX_DESCRIPTION_LENGTH = 2000; + +export interface ErrorConflict { + value: string; + count: number; + objects: string[]; +} + +export interface ErrorDetails { + source?: string; + httpStatus?: number; + description?: string; + conflicts?: ErrorConflict[]; + totalConflicts?: number; + fullMessage?: string; + [key: string]: unknown; +} + +/** + * Throw from a handler when the failure has structured context worth keeping. + * Keep `message` short and stable (it is grouped on in the top-errors view); put + * the variable parts in `details`. + */ +export class StepError extends Error { + readonly details: ErrorDetails; + + constructor(message: string, details: ErrorDetails, options?: ErrorOptions) { + super(message, options); + this.name = "StepError"; + this.details = details; + } +} + +/** Collapse repeated conflicts by message, keeping counts and a sample of objects. */ +export function groupConflicts( + conflicts: Array<{ object?: string; value: string }>, + sampleSize = 5 +): ErrorConflict[] { + const groups = new Map(); + for (const { object, value } of conflicts) { + const group = groups.get(value) ?? { value, count: 0, objects: [] }; + group.count += 1; + if (object && group.objects.length < sampleSize && !group.objects.includes(object)) { + group.objects.push(object); + } + groups.set(value, group); + } + return [...groups.values()].sort((a, b) => b.count - a.count); +} + +function truncate(text: string, max: number): string { + return text.length > max ? `${text.slice(0, max - 1)}…` : text; +} + +function describeResponseBody(data: unknown): string | undefined { + if (data == null || data === "") { + return undefined; + } + if (typeof data === "string") { + return truncate(data, MAX_DESCRIPTION_LENGTH); + } + if (typeof data === "object") { + const body = data as Record; + for (const key of ["message", "detail", "description", "error"]) { + if (typeof body[key] === "string" && body[key]) { + return truncate(body[key], MAX_DESCRIPTION_LENGTH); + } + } + } + return truncate(JSON.stringify(data), MAX_DESCRIPTION_LENGTH); +} + +function axiosDetails(err: AxiosError): ErrorDetails { + const method = err.config?.method?.toUpperCase(); + const url = err.config?.url; + return { + httpStatus: err.response?.status, + request: method && url ? `${method} ${url}` : url, + description: describeResponseBody(err.response?.data), + }; +} + +/** + * Split an error into the columns persisted on StepExecution/TaskExecution: + * a short headline, the stack, and structured details (if any). + */ +export function toErrorRecord(err: Error): { + errorMessage: string; + errorStack: string | undefined; + errorDetails: ErrorDetails | undefined; +} { + let details: ErrorDetails | undefined; + if (err instanceof StepError) { + details = { ...err.details }; + } else if (err instanceof AxiosError) { + details = axiosDetails(err); + } + + const errorMessage = truncate(err.message, MAX_ERROR_MESSAGE_LENGTH); + if (errorMessage !== err.message) { + details = { ...details, fullMessage: err.message }; + } + + return { errorMessage, errorStack: err.stack, errorDetails: details }; +} diff --git a/src/tests/unit/errorRecord.test.ts b/src/tests/unit/errorRecord.test.ts new file mode 100644 index 0000000..44ca6df --- /dev/null +++ b/src/tests/unit/errorRecord.test.ts @@ -0,0 +1,68 @@ +import { describe, it, expect } from "vitest"; +import { AxiosError, AxiosHeaders } from "axios"; +import { + groupConflicts, + MAX_ERROR_MESSAGE_LENGTH, + StepError, + toErrorRecord, +} from "@/shared/utils/error.ts"; + +describe("groupConflicts", () => { + it("collapses repeated values, counts them and samples objects", () => { + const conflicts = [ + ...Array.from({ length: 8 }, (_, i) => ({ object: `ou${i}`, value: "Org unit not found" })), + { object: "de1", value: "Data element not found" }, + ]; + + expect(groupConflicts(conflicts)).toEqual([ + { value: "Org unit not found", count: 8, objects: ["ou0", "ou1", "ou2", "ou3", "ou4"] }, + { value: "Data element not found", count: 1, objects: ["de1"] }, + ]); + }); +}); + +describe("toErrorRecord", () => { + it("keeps short plain errors as-is with no details", () => { + const record = toErrorRecord(new Error("boom")); + expect(record.errorMessage).toBe("boom"); + expect(record.errorDetails).toBeUndefined(); + }); + + it("truncates long messages and keeps the original in details", () => { + const long = "x".repeat(MAX_ERROR_MESSAGE_LENGTH + 50); + const record = toErrorRecord(new Error(long)); + expect(record.errorMessage).toHaveLength(MAX_ERROR_MESSAGE_LENGTH); + expect(record.errorMessage.endsWith("…")).toBe(true); + expect(record.errorDetails?.fullMessage).toBe(long); + }); + + it("carries StepError details through", () => { + const record = toErrorRecord( + new StepError("Upstream failed", { source: "dhis2", totalConflicts: 3 }) + ); + expect(record.errorMessage).toBe("Upstream failed"); + expect(record.errorDetails).toEqual({ source: "dhis2", totalConflicts: 3 }); + }); + + it("extracts status, request and upstream message from axios errors", () => { + const err = new AxiosError( + "Request failed with status code 500", + "ERR_BAD_RESPONSE", + { method: "post", url: "/predictions", headers: new AxiosHeaders() }, + undefined, + { + status: 500, + statusText: "Internal Server Error", + headers: {}, + config: { headers: new AxiosHeaders() }, + data: { detail: "Model crashed" }, + } + ); + + expect(toErrorRecord(err).errorDetails).toEqual({ + httpStatus: 500, + request: "POST /predictions", + description: "Model crashed", + }); + }); +}); diff --git a/src/tests/unit/predictionDataDownload.test.ts b/src/tests/unit/predictionDataDownload.test.ts index a44969a..4a66eff 100644 --- a/src/tests/unit/predictionDataDownload.test.ts +++ b/src/tests/unit/predictionDataDownload.test.ts @@ -19,6 +19,7 @@ import { getPredictionResult, getPredictionSetup } from "@/services/worker/utils import { predictionDataDownload } from "@/services/worker/services/handlers/predictionDataDownload/index.ts"; import { dhis2DataUpload } from "@/services/worker/services/handlers/dhis2DataUpload/index.ts"; import { postDataValueSet } from "@/services/worker/utils/dhis2.ts"; +import { StepError } from "@/shared/utils/error.ts"; vi.mock("@/services/worker/utils/dhis2.ts", () => ({ postDataValueSet: vi.fn(), @@ -240,7 +241,15 @@ describe("dhis2DataUpload filename resolution", () => { }); const ctx = buildMockContext({ input: { filename: "f.json" }, handlerConfig: {} }); - await expect(dhis2DataUpload.execute(ctx)).rejects.toThrow(/de: Data element not found/); + const error = await dhis2DataUpload.execute(ctx).catch((e: unknown) => e); + + expect(error).toBeInstanceOf(StepError); + expect((error as StepError).message).toBe("DHIS2 rejected the data value import"); + expect((error as StepError).details).toMatchObject({ + source: "dhis2", + totalConflicts: 1, + conflicts: [{ value: "Data element not found", count: 1, objects: ["de"] }], + }); }); it("returns real counts and logs conflicts on WARNING", async () => { @@ -260,7 +269,10 @@ describe("dhis2DataUpload filename resolution", () => { expect(ctx.log).toHaveBeenCalledWith( "WARN", expect.stringMatching(/conflicts/), - expect.objectContaining({ conflicts: ["ou: Org unit not in hierarchy"] }) + expect.objectContaining({ + totalConflicts: 1, + conflicts: [{ value: "Org unit not in hierarchy", count: 1, objects: ["ou"] }], + }) ); }); diff --git a/src/tests/utils/helpers.ts b/src/tests/utils/helpers.ts index 1dc29e1..75d3681 100644 --- a/src/tests/utils/helpers.ts +++ b/src/tests/utils/helpers.ts @@ -73,6 +73,7 @@ export function buildMockTaskExecution(overrides?: Partial): Task output: null, errorMessage: null, errorStack: null, + errorDetails: null, startedAt: new Date(), finishedAt: null, createdAt: new Date(), @@ -129,6 +130,7 @@ export function buildMockStepExecution(overrides?: Partial): Step output: null, errorMessage: null, errorStack: null, + errorDetails: null, resultProcessedAt: null, startedAt: new Date(), finishedAt: null, From 5e26683d70c79fda23bea99a6546efc6af6550a3 Mon Sep 17 00:00:00 2001 From: Allen7702 Date: Sat, 26 Sep 2026 12:59:26 +0300 Subject: [PATCH 4/5] chore: add climate dataset freshness handler to the existing handlers enum --- CLAUDE.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CLAUDE.md b/CLAUDE.md index 3e15edc..85faea7 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -133,7 +133,7 @@ interface StepHandler { `StepContext` (`src/services/worker/types/service.ts`) provides the step definition, shared pipeline context (JSON passed between steps), sub-task reporting (`ctx.tasks`), and a `ctx.log()` method that writes to `ExecutionLog`. `src/shared/handlers/catalog.ts` converts each handler's Zod config/context schemas to JSON Schema for the API's handler-catalog endpoint. -Current handlers (see `Handlers` enum): climate acquisition via openEO (`climate-openeo-{create,poll,download}`), CHAP predictions (`prediction-{trigger,poll,data-download}`), and DHIS2 import (`dhis2-data-upload`, `dhis2-analytics-run`, `threshold-generation`, `alert-generation`, `event-data-upload`). +Current handlers (see `Handlers` enum): climate dataset freshness (`open-climate-service-sync`), climate acquisition via openEO (`climate-openeo-{create,poll,download}`), CHAP predictions (`prediction-{trigger,poll,data-download}`), and DHIS2 import (`dhis2-data-upload`, `dhis2-analytics-run`, `threshold-generation`, `alert-generation`, `event-data-upload`). ### Key Files From 1140d2499aeb9b6dd55c5705a1dc991ce67edf71 Mon Sep 17 00:00:00 2001 From: Allen7702 Date: Sun, 27 Sep 2026 09:35:40 +0300 Subject: [PATCH 5/5] chore: update version to 1.3.0 in package.json --- package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/package.json b/package.json index 3c2fae0..62864f1 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "caps-engine", "module": "src/index.ts", - "version": "1.0.0", + "version": "1.3.0", "license": "BSD-3-Clause", "type": "module", "private": true,