Feature/ocs sync

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
 

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,

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;

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?

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'

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<Handlers, HandlerRegistryEntry> = 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<Handlers, HandlerRegistryEntry> = new Map<
       displayName: "DHIS2 Data Upload",
       description: "Uploads data from a file to DHIS2",
       tags: ["dhis2"],
-      schemas: {},
+      schemas: {
+        config: dhis2DataUploadConfigSchema,
+      },
     },
   ],
   [

src/services/worker/services/handlers/dhis2DataUpload/index.ts

@@ -4,7 +4,8 @@ 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 { groupConflicts, StepError } from "@/shared/utils/error.ts";
+import { dhis2DataUploadConfigSchema } from "@/services/worker/services/handlers/dhis2DataUpload/schemas/config.ts";
 
 /**
  * Handler: dhis2-data-upload
@@ -15,11 +16,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 +44,7 @@ export const dhis2DataUpload: StepHandler = {
 
     const task = await ctx.tasks.startTask("upload-data-value-set", {
       filePath,
+      importStrategy,
     });
 
     let payload: DataValueSet;
@@ -58,36 +68,39 @@ 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 ?? [];
+
     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}` : ""}`
-      );
+      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;
     }
 
+    if (conflicts.length > 0) {
+      await ctx.log("WARN", "DHIS2 data value import completed with conflicts", {
+        status: summary.status,
+        totalConflicts: conflicts.length,
+        conflicts: groupConflicts(conflicts),
+      });
+    }
+
     const counts = summary.importCount ?? {
       imported: 0,
       updated: 0,

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<typeof dataValueImportStrategySchema>;
+
+export const dhis2DataUploadConfigSchema = z.object({
+  importStrategy: dataValueImportStrategySchema.default("CREATE_AND_UPDATE"),
+});
+
+export type Dhis2DataUploadConfig = z.infer<typeof dhis2DataUploadConfigSchema>;

src/services/worker/services/handlers/index.ts

@@ -1,3 +1,4 @@
+export * from "./openClimateServiceSync";
 export * from "./climateOpenEoCreate";
 export * from "./climateOpenEoPoll";
 export * from "./climateOpenEoDownload";