Gate Open Dataset Loads on CKAN Schema, Licence and Row-Count Drift with Gemini Impact Notes and Human Approval

id: ckan-open-dataset-schema-and-licence-drift-gate
namespace: company.team

description: |
  Watch a dataset published on a CKAN open-data portal (data.gov.au, data.qld.gov.au and
  hundreds of other national and city portals) and stop downstream loads before they break.
  Each run reads the live DataStore schema, licence and row count, diffs them against
  the snapshot saved by the previous run, lets Gemini explain what a breaking change
  means for consumers, pauses for a human decision only when the change is breaking,
  and appends every check to a DuckDB ledger so you can see how often the publisher
  changes the dataset under you.

inputs:
  - id: portal_url
    type: STRING
    defaults: https://data.gov.au/data
    description: Base URL of the CKAN portal, without a trailing slash. Any CKAN 2.x
      portal whose DataStore API answers anonymously works (data.gov.au,
      www.data.qld.gov.au, ...); portals that require a token need an
      Authorization header, see the blueprint notes.

  - id: dataset_id
    type: STRING
    defaults: energy-rating-data-for-household-appliances-non-labelled-products
    description: CKAN dataset (package) name or id, the last path segment of the
      dataset page URL.

  - id: resource_id
    type: STRING
    required: false
    description: Optional resource (file) id inside the dataset. Leave empty to
      check the first resource that is loaded into the portal's DataStore.

  - id: row_count_tolerance
    type: FLOAT
    defaults: 0.2
    description: Relative change in row count that counts as breaking (0.2 = a drop
      or jump of more than 20 percent). Open datasets grow; they rarely shrink
      by a fifth without a reason.

  - id: model_name
    type: STRING
    defaults: gemini-2.5-flash
    description: Google Gemini model that explains the impact of a breaking change.

triggers:
  - id: daily_check
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 6 * * *"
    description: Check the dataset every day at 06:00 UTC, before the pipelines that
      consume it run.

tasks:
  - id: package_show
    type: io.kestra.plugin.core.http.Request
    description: Dataset metadata from the CKAN Action API (keyless), including
      title, licence, last modification and the list of resources with their
      DataStore status.
    uri: "{{ inputs.portal_url }}/api/3/action/package_show?id={{ inputs.dataset_id
      | urlencode }}"
    headers:
      Accept: application/json

  - id: pick_resource
    type: io.kestra.plugin.graalvm.js.Eval
    description: Picks the resource to watch (the one given as input, or the first
      one loaded into the DataStore) and lifts the dataset-level facts that can
      drift (licence, modification date).
    outputs:
      - resource_id
      - resource_name
      - dataset_title
      - license_id
      - license_title
      - metadata_modified
      - resource_last_modified
    script: |
      // The value of the last expression is the task output; declared `outputs` are picked from its keys.
      (function () {
        const body = JSON.parse({{ outputs.package_show.body | toJson }});
        const wanted = {{ (inputs.resource_id ?? '') | toJson }};
        if (!body.success) throw new Error('package_show failed: ' + JSON.stringify(body.error));
        const pkg = body.result;
        const resources = pkg.resources || [];
        const resource = wanted
          ? resources.find((r) => r.id === wanted)
          : resources.find((r) => r.datastore_active === true || r.datastore_active === 'true');
        if (!resource) {
          throw new Error(wanted
            ? 'Resource ' + wanted + ' is not part of dataset ' + pkg.name
            : 'Dataset ' + pkg.name + ' has no resource loaded into the DataStore; pass resource_id or pick another dataset');
        }
        logger.info('Watching "' + String(resource.name || resource.id).replace(/\s+/g, ' ').trim() + '" (' + resource.id + ') in "' + pkg.title + '", licence ' + pkg.license_id);
        return {
          resource_id: resource.id,
          resource_name: String(resource.name || resource.id).replace(/\s+/g, ' ').trim(),
          dataset_title: pkg.title || pkg.name,
          license_id: pkg.license_id || '',
          license_title: pkg.license_title || '',
          metadata_modified: pkg.metadata_modified || '',
          resource_last_modified: resource.last_modified || ''
        };
      })()

  - id: datastore_fields
    type: io.kestra.plugin.core.http.Request
    description: The DataStore's own view of the resource with limit=0, which
      returns the typed column list and the total row count without downloading
      any rows.
    uri: "{{ inputs.portal_url }}/api/3/action/datastore_search?resource_id={{
      outputs.pick_resource.outputs.resource_id }}&limit=0"
    headers:
      Accept: application/json

  - id: previous_snapshot
    type: io.kestra.plugin.core.kv.Get
    key: "ckan_snapshot_{{ outputs.pick_resource.outputs.resource_id }}"
    errorOnMissing: false
    description: Snapshot (columns, types, licence, row count) saved by the last run
      that was approved or clean. Missing on the first run, which then only
      records a baseline.

  - id: diff_schema
    type: io.kestra.plugin.graalvm.js.Eval
    description: |
      Deterministic diff between the live dataset and the saved snapshot. Removed or retyped
      columns, a licence change or a row-count change beyond the tolerance are BREAKING;
      added columns or a new publication date are INFO; an identical dataset is NONE; the
      first run is BASELINE. Arrays are passed as JSON text because the GraalVM bridge
      hands JS arrays to Kestra as index-keyed maps.
    outputs:
      - severity
      - column_count
      - row_count
      - previous_row_count
      - added_json
      - removed_json
      - retyped_json
      - license_changed
      - previous_license
      - changes_json
      - snapshot_json
    script: |
      (function () {
        const body = JSON.parse({{ outputs.datastore_fields.body | toJson }});
        const meta = {{ outputs.pick_resource.outputs | toJson }};
        const prevRaw = {{ (outputs.previous_snapshot.value ?? '') | toJson }};
        const tolerance = {{ inputs.row_count_tolerance }};
        if (!body.success) throw new Error('datastore_search failed: ' + JSON.stringify(body.error));

        const prev = typeof prevRaw === 'string' ? (prevRaw ? JSON.parse(prevRaw) : null) : prevRaw;
        const fields = {};
        for (const f of body.result.fields || []) {
          if (f.id !== '_id') fields[f.id] = f.type;
        }
        const rowCount = body.result.total == null ? null : Number(body.result.total);
        const snapshot = {
          fields: fields,
          row_count: rowCount,
          license_id: meta.license_id,
          license_title: meta.license_title,
          metadata_modified: meta.metadata_modified,
          resource_last_modified: meta.resource_last_modified,
          captured_at: new Date().toISOString(),
          execution_id: {{ execution.id | toJson }}
        };

        const added = [], removed = [], retyped = [], changes = [];
        let severity = 'BASELINE';
        let licenseChanged = false;
        let prevRows = null;
        if (prev && prev.fields) {
          severity = 'NONE';
          const prevFields = prev.fields;
          for (const name of Object.keys(fields)) {
            if (!(name in prevFields)) added.push({ column: name, type: fields[name] });
            else if (prevFields[name] !== fields[name]) retyped.push({ column: name, from: prevFields[name], to: fields[name] });
          }
          for (const name of Object.keys(prevFields)) {
            if (!(name in fields)) removed.push({ column: name, type: prevFields[name] });
          }
          licenseChanged = (prev.license_id || '') !== (meta.license_id || '');
          prevRows = prev.row_count == null ? null : Number(prev.row_count);

          if (removed.length) changes.push('BREAKING: ' + removed.length + ' column(s) removed: ' + removed.map((c) => c.column).join(', '));
          if (retyped.length) changes.push('BREAKING: ' + retyped.length + ' column(s) retyped: ' + retyped.map((c) => c.column + ' ' + c.from + ' -> ' + c.to).join(', '));
          if (licenseChanged) changes.push('BREAKING: licence changed from ' + (prev.license_id || 'none') + ' to ' + (meta.license_id || 'none'));
          if (prevRows != null && rowCount != null) {
            const delta = Math.abs(rowCount - prevRows) / Math.max(prevRows, 1);
            if (delta > tolerance) changes.push('BREAKING: row count moved from ' + prevRows + ' to ' + rowCount + ' (' + Math.round(delta * 100) + ' percent, tolerance ' + Math.round(tolerance * 100) + ' percent)');
            else if (rowCount !== prevRows) changes.push('INFO: row count moved from ' + prevRows + ' to ' + rowCount);
          }
          if (added.length) changes.push('INFO: ' + added.length + ' column(s) added: ' + added.map((c) => c.column).join(', '));
          if ((prev.metadata_modified || '') !== (meta.metadata_modified || '')) changes.push('INFO: dataset metadata modified ' + prev.metadata_modified + ' -> ' + meta.metadata_modified);

          if (changes.some((c) => c.startsWith('BREAKING'))) severity = 'BREAKING';
          else if (changes.length) severity = 'INFO';
        } else {
          changes.push('BASELINE: first snapshot, ' + Object.keys(fields).length + ' columns, ' + rowCount + ' rows, licence ' + (meta.license_id || 'none'));
        }

        logger.info(severity + ' for ' + meta.dataset_title + ' / ' + meta.resource_name + ': ' + (changes.join(' | ') || 'no change'));
        return {
          severity: severity,
          column_count: Object.keys(fields).length,
          row_count: rowCount,
          previous_row_count: prevRows,
          added_json: JSON.stringify(added),
          removed_json: JSON.stringify(removed),
          retyped_json: JSON.stringify(retyped),
          license_changed: licenseChanged,
          previous_license: prev ? (prev.license_id || '') : '',
          changes_json: JSON.stringify(changes),
          snapshot_json: JSON.stringify(snapshot)
        };
      })()

  - id: breaking_gate
    type: io.kestra.plugin.core.flow.If
    description: Human in the loop only for breaking changes. Clean runs and
      additive changes pass without anyone being paged.
    condition: "{{ outputs.diff_schema.outputs.severity == 'BREAKING' }}"
    then:
      - id: explain_impact
        type: io.kestra.plugin.ai.completion.ChatCompletion
        allowFailure: true
        description: |
          Turns the raw diff into a note a data owner can act on: which consumers break, whether
          the licence change restricts redistribution, and the recommended action. allowFailure:
          when the model call fails (missing key, quota) the pause still happens, with the raw diff only.
        provider:
          type: io.kestra.plugin.ai.provider.GoogleGemini
          apiKey: "{{ secret('GEMINI_API_KEY') }}"
          modelName: "{{ inputs.model_name }}"
        configuration:
          temperature: 0.0
          responseFormat:
            type: JSON
            jsonSchema:
              type: object
              properties:
                summary:
                  type: string
                affected_columns:
                  type: array
                  items:
                    type: string
                recommended_action:
                  type: string
                  enum: [ UPDATE_LOADER, REVIEW_LICENCE, CONTACT_PUBLISHER, SAFE_TO_APPROVE ]
                rationale:
                  type: string
              required: [ summary, affected_columns, recommended_action, rationale ]
        messages:
          - type: SYSTEM
            content: |
              You are a data engineer who maintains pipelines that load open government datasets
              from CKAN portals into a warehouse. You receive the detected changes between the
              previously approved version of a dataset and the version published now. Explain in
              at most 60 words what breaks for a typical loader (removed columns break SELECT
              lists and joins, retyped columns break casts, a licence change may restrict reuse,
              a large row-count drop usually means a partial publication). List the affected
              columns, then recommend exactly one action: UPDATE_LOADER (schema change the
              consumer must adopt), REVIEW_LICENCE (licence terms changed), CONTACT_PUBLISHER
              (the publication looks incomplete or wrong) or SAFE_TO_APPROVE. Use only the
              changes provided; never invent columns.
          - type: USER
            content: |
              Portal: {{ inputs.portal_url }}
              Dataset: {{ outputs.pick_resource.outputs.dataset_title }} ({{ inputs.dataset_id }})
              Resource: {{ outputs.pick_resource.outputs.resource_name }}
              Detected changes:
              {{ outputs.diff_schema.outputs.changes_json }}
              Removed columns: {{ outputs.diff_schema.outputs.removed_json }}
              Retyped columns: {{ outputs.diff_schema.outputs.retyped_json }}
              Licence: {{ outputs.diff_schema.outputs.previous_license }} -> {{ outputs.pick_resource.outputs.license_id }} ({{ outputs.pick_resource.outputs.license_title }})

      - id: notify_owner
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        allowFailure: true
        description: Posts the change, the model's note and a deep link to this
          execution. allowFailure keeps the pause reachable when the webhook is
          not configured.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": ":warning: Breaking change in open dataset *{{ outputs.pick_resource.outputs.dataset_title }}* ({{ outputs.pick_resource.outputs.resource_name }})\n{{ outputs.diff_schema.outputs.changes_json }}\nAI: {{ outputs.explain_impact.jsonOutput.summary ?? 'model unavailable' }} ({{ outputs.explain_impact.jsonOutput.recommended_action ?? 'n/a' }})\nDecide here: {{ kestra.url ?? '' }}/ui/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}"
          }

      - id: approve_change
        type: io.kestra.plugin.core.flow.Pause
        description: Waits for the data owner. approve moves the baseline to the new
          schema (after the loader is fixed); hold keeps the old baseline so the
          next run alerts again until the publisher or the loader is fixed.
        onResume:
          - id: decision
            type: SELECT
            displayName: Decision
            description: approve = accept the new schema and licence as the baseline; hold =
              keep the previous baseline and keep alerting
            values: [ approve, hold ]
            defaults: hold
          - id: note
            type: STRING
            displayName: Note (ticket, reason, publisher contact)
            required: false
    else:
      - id: no_action_needed
        type: io.kestra.plugin.core.log.Log
        message: "{{ outputs.diff_schema.outputs.severity }}: {{
          outputs.diff_schema.outputs.changes_json }}"

  - id: finalize_decision
    type: io.kestra.plugin.graalvm.js.Eval
    description: Resolves the decision (system for BASELINE, NONE and INFO; the
      reviewer's choice for BREAKING) and whether the snapshot moves.
    outputs:
      - decision
      - decided_by
      - update_snapshot
      - ai_summary
      - ai_action
    script: |
      (function () {
        const severity = {{ outputs.diff_schema.outputs.severity | toJson }};
        const resume = {{ (outputs.approve_change.onResume ?? {}) | toJson }};
        const ai = {{ (outputs.explain_impact.jsonOutput ?? {}) | toJson }};
        let decision, decidedBy;
        if (severity === 'BASELINE') { decision = 'baseline'; decidedBy = 'system'; }
        else if (severity === 'BREAKING') { decision = resume.decision || 'hold'; decidedBy = 'human'; }
        else { decision = 'auto_pass'; decidedBy = 'system'; }
        return {
          decision: decision,
          decided_by: decidedBy,
          update_snapshot: decision !== 'hold',
          ai_summary: ai.summary || (severity === 'BREAKING' ? 'UNAVAILABLE' : ''),
          ai_action: ai.recommended_action || ''
        };
      })()

  - id: previous_ledger
    type: io.kestra.plugin.core.kv.Get
    key: ckan_drift_ledger_uri
    errorOnMissing: false
    description: Kestra's KV store holds the internal-storage URI of the DuckDB
      ledger written by the previous execution (shared by every dataset watched
      with this flow).

  - id: write_ledger
    type: io.kestra.plugin.core.flow.If
    description: Append this check to the DuckDB ledger (created on first run) and
      return how often this dataset has changed and how often a human had to
      step in.
    condition: "{{ outputs.previous_ledger.value is defined and
      outputs.previous_ledger.value is not empty }}"
    then:
      - id: append_event
        type: io.kestra.plugin.jdbc.duckdb.Queries
        databaseUri: "{{ outputs.previous_ledger.value }}"
        outputDbFile: true
        fetchType: FETCH_ONE
        parameters:
          execution_id: "{{ execution.id }}"
          portal_url: "{{ inputs.portal_url }}"
          dataset_id: "{{ inputs.dataset_id }}"
          resource_id: "{{ outputs.pick_resource.outputs.resource_id }}"
          severity: "{{ outputs.diff_schema.outputs.severity }}"
          column_count: "{{ outputs.diff_schema.outputs.column_count }}"
          row_count: "{{ outputs.diff_schema.outputs.row_count ?? '' }}"
          previous_row_count: "{{ outputs.diff_schema.outputs.previous_row_count ?? '' }}"
          license_id: "{{ outputs.pick_resource.outputs.license_id }}"
          license_changed: "{{ outputs.diff_schema.outputs.license_changed }}"
          changes: "{{ outputs.diff_schema.outputs.changes_json }}"
          decision: "{{ outputs.finalize_decision.outputs.decision }}"
          decided_by: "{{ outputs.finalize_decision.outputs.decided_by }}"
          ai_summary: "{{ outputs.finalize_decision.outputs.ai_summary }}"
          ai_action: "{{ outputs.finalize_decision.outputs.ai_action }}"
          reviewer_note: "{{ outputs.approve_change.onResume.note ?? '' }}"
        sql: |
          INSERT INTO drift_events (execution_id, portal_url, dataset_id, resource_id, severity, column_count, row_count, previous_row_count, license_id, license_changed, changes, decision, decided_by, ai_summary, ai_action, reviewer_note)
          VALUES (:execution_id, :portal_url, :dataset_id, :resource_id, :severity, CAST(:column_count AS INTEGER), CAST(NULLIF(:row_count, '') AS BIGINT), CAST(NULLIF(:previous_row_count, '') AS BIGINT), :license_id, CAST(:license_changed AS BOOLEAN), :changes, :decision, :decided_by, :ai_summary, :ai_action, :reviewer_note);
          SELECT
            count(*) AS checks,
            count(*) FILTER (WHERE severity = 'BREAKING') AS breaking,
            count(*) FILTER (WHERE severity = 'INFO') AS additive,
            count(*) FILTER (WHERE decided_by = 'human') AS human_decisions,
            count(*) FILTER (WHERE decision = 'hold') AS holds,
            count(DISTINCT resource_id) AS resources_watched
          FROM drift_events
          WHERE dataset_id = :dataset_id;
    else:
      - id: create_ledger
        type: io.kestra.plugin.jdbc.duckdb.Queries
        outputDbFile: true
        fetchType: FETCH_ONE
        parameters:
          execution_id: "{{ execution.id }}"
          portal_url: "{{ inputs.portal_url }}"
          dataset_id: "{{ inputs.dataset_id }}"
          resource_id: "{{ outputs.pick_resource.outputs.resource_id }}"
          severity: "{{ outputs.diff_schema.outputs.severity }}"
          column_count: "{{ outputs.diff_schema.outputs.column_count }}"
          row_count: "{{ outputs.diff_schema.outputs.row_count ?? '' }}"
          previous_row_count: "{{ outputs.diff_schema.outputs.previous_row_count ?? '' }}"
          license_id: "{{ outputs.pick_resource.outputs.license_id }}"
          license_changed: "{{ outputs.diff_schema.outputs.license_changed }}"
          changes: "{{ outputs.diff_schema.outputs.changes_json }}"
          decision: "{{ outputs.finalize_decision.outputs.decision }}"
          decided_by: "{{ outputs.finalize_decision.outputs.decided_by }}"
          ai_summary: "{{ outputs.finalize_decision.outputs.ai_summary }}"
          ai_action: "{{ outputs.finalize_decision.outputs.ai_action }}"
          reviewer_note: "{{ outputs.approve_change.onResume.note ?? '' }}"
        sql: |
          CREATE TABLE drift_events (
            execution_id VARCHAR, checked_at TIMESTAMP DEFAULT now(), portal_url VARCHAR, dataset_id VARCHAR, resource_id VARCHAR,
            severity VARCHAR, column_count INTEGER, row_count BIGINT, previous_row_count BIGINT, license_id VARCHAR, license_changed BOOLEAN,
            changes JSON, decision VARCHAR, decided_by VARCHAR, ai_summary VARCHAR, ai_action VARCHAR, reviewer_note VARCHAR
          );
          INSERT INTO drift_events (execution_id, portal_url, dataset_id, resource_id, severity, column_count, row_count, previous_row_count, license_id, license_changed, changes, decision, decided_by, ai_summary, ai_action, reviewer_note)
          VALUES (:execution_id, :portal_url, :dataset_id, :resource_id, :severity, CAST(:column_count AS INTEGER), CAST(NULLIF(:row_count, '') AS BIGINT), CAST(NULLIF(:previous_row_count, '') AS BIGINT), :license_id, CAST(:license_changed AS BOOLEAN), :changes, :decision, :decided_by, :ai_summary, :ai_action, :reviewer_note);
          SELECT
            count(*) AS checks,
            count(*) FILTER (WHERE severity = 'BREAKING') AS breaking,
            count(*) FILTER (WHERE severity = 'INFO') AS additive,
            count(*) FILTER (WHERE decided_by = 'human') AS human_decisions,
            count(*) FILTER (WHERE decision = 'hold') AS holds,
            count(DISTINCT resource_id) AS resources_watched
          FROM drift_events
          WHERE dataset_id = :dataset_id;

  - id: save_ledger_uri
    type: io.kestra.plugin.core.kv.Set
    key: ckan_drift_ledger_uri
    value: "{{ (outputs.append_event ?? outputs.create_ledger).databaseUri }}"

  - id: save_snapshot
    type: io.kestra.plugin.core.kv.Set
    runIf: "{{ outputs.finalize_decision.outputs.update_snapshot }}"
    description: Moves the baseline to what was observed now. Skipped on hold, so
      the next run compares against the last approved version and alerts again.
    key: "ckan_snapshot_{{ outputs.pick_resource.outputs.resource_id }}"
    value: "{{ outputs.diff_schema.outputs.snapshot_json }}"

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: |
      {{ outputs.pick_resource.outputs.dataset_title }} / {{ outputs.pick_resource.outputs.resource_name }}: {{ outputs.diff_schema.outputs.severity }}, decision {{ outputs.finalize_decision.outputs.decision }} by {{ outputs.finalize_decision.outputs.decided_by }}.
      {{ outputs.diff_schema.outputs.column_count }} columns, {{ outputs.diff_schema.outputs.row_count }} rows, licence {{ outputs.pick_resource.outputs.license_id }}.
      Changes: {{ outputs.diff_schema.outputs.changes_json }}
      Ledger: {{ (outputs.append_event ?? outputs.create_ledger).outputs[0].row | toJson }}

outputs:
  - id: severity
    type: STRING
    value: "{{ outputs.diff_schema.outputs.severity }}"
  - id: decision
    type: STRING
    value: "{{ outputs.finalize_decision.outputs.decision }}"
  - id: changes
    type: JSON
    value: "{{ outputs.diff_schema.outputs.changes_json }}"
  - id: ledger_summary
    type: JSON
    value: "{{ (outputs.append_event ?? outputs.create_ledger).outputs[0].row }}"