UI Test Flow for SQL and Kafka Job Executions

test-flow/README.md

@@ -122,7 +122,7 @@ Tool-only by default. Options:
 
 ```bash
 test-flow/run-manual-flow.sh                 # tool-only
-test-flow/run-manual-flow.sh --only NAME     # one behavior: plugins|batch|archive|streaming|scheduling|kafka
+test-flow/run-manual-flow.sh --only NAME     # one behavior: plugins|batch|archive|streaming|scheduling|kafka|sql
 test-flow/run-manual-flow.sh --with-web      # add the web UI (visual), seeded with real projects
 test-flow/run-manual-flow.sh --with-efk      # add the Kibana "Executions" dashboard
 test-flow/run-manual-flow.sh --down          # stop and remove everything
@@ -132,6 +132,26 @@ Visual checks you can do with `--with-web` (not automated): import a REDCap data
 schemas, open the REDCap page, view the Executions dashboard (`--with-efk`), and browse the mapped
 data in the FHIR server.
 
+### SQL and Kafka source jobs (run them from the web UI)
+
+The stack also brings up a **Postgres** container as a SQL data source, seeded on start from
+[`data/sql-source-seed.sql`](data/sql-source-seed.sql) (a small `patients` table). Two ways to exercise
+the SQL/Kafka sources:
+
+- **Headless behavior:** `run-manual-flow.sh --only sql` maps the Postgres `patients` table to FHIR,
+  alongside the existing `--only kafka` (which publishes `data/redcap-patients.ndjson` to the
+  `redcap-patients` topic). Both also run as part of a full run.
+- **From the UI:** with the stack up, [`create-ui-testflow.sh`](create-ui-testflow.sh) creates a
+  `test-flow` project through the REST API — a Patient schema + mapping and two jobs, one SQL (Postgres
+  `patients`) and one Kafka (`redcap-patients`) — so both appear in the web UI and can be run and
+  observed there, with executions in Kibana. The jobs connect to `jdbc:postgresql://postgres:5432/ignifyr`
+  (user/pass `ignifyr`) and topic `redcap-patients`.
+
+Caveat for the Kafka job: the mapping produces deterministic Patient ids, so re-publishing the *same*
+records into a live topic makes one micro-batch contain duplicate ids, which the FHIR server rejects
+(`Resource identity overlapping`). Start clean with `--down` (the topic is ephemeral) or publish each
+record once; distinct records don't collide.
+
 ---
 
 ## Notes

test-flow/config/jobs/sql-patient-job.json

@@ -0,0 +1,33 @@
+{
+  "id": "sql-patient-job",
+  "sourceSettings": {
+    "source": {
+      "jsonClass": "SqlSourceSettings",
+      "name": "sql-source",
+      "sourceUri": "https://ignifyr.io/test-flow/sql",
+      "databaseUrl": "jdbc:postgresql://postgres:5432/ignifyr",
+      "username": "ignifyr",
+      "password": "ignifyr"
+    }
+  },
+  "sinkSettings": {
+    "jsonClass": "FhirRepositorySinkSettings",
+    "fhirRepoUrl": "http://repofyr:8080/fhir"
+  },
+  "mappings": [
+    {
+      "name": "patient-mapping",
+      "mappingRef": "https://aiccelerate.eu/fhir/mappings/patient-mapping",
+      "sourceBinding": {
+        "source": {
+          "jsonClass": "SqlSource",
+          "tableName": "patients"
+        }
+      }
+    }
+  ],
+  "dataProcessingSettings": {
+    "saveErroneousRecords": false,
+    "archiveMode": "off"
+  }
+}

test-flow/create-ui-testflow.sh

@@ -0,0 +1,156 @@
+#!/usr/bin/env bash
+#
+# Create a "test-flow" project in the RUNNING ignifyr-server via its REST API, so a SQL job (Postgres
+# 'patients' table) and a Kafka job (redcap-patients topic) show up in the web UI and can be run and
+# observed from there. Idempotent-ish: re-running reports 409 (already exists) and continues.
+#
+# Prereq: the manual stack is up (test-flow/run-manual-flow.sh --with-web [--with-efk]) so the server,
+# Postgres and Kafka containers are running. Needs curl.
+#
+# Usage: test-flow/create-ui-testflow.sh
+set -uo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
+BASE="http://localhost:8085/ignifyr"
+PROJECT="test-flow"
+
+SCHEMA_FILE="$REPO_ROOT/ignifyr-testkit/src/main/resources/test-schemas/some-folder-1/Ext-patient.StructureDefinition.json"
+MAPPING_FILE="$REPO_ROOT/ignifyr-testkit/src/main/resources/test-mappings/some-folder-1/patient-mapping.json"
+TMP="$(mktemp -d)"; trap 'rm -rf "$TMP"' EXIT
+FAILC=0
+
+command -v curl >/dev/null 2>&1 || { echo "curl required" >&2; exit 1; }
+curl -fsS -X OPTIONS "$BASE/projects" >/dev/null 2>&1 || { echo "ignifyr-server not reachable at $BASE — bring the stack up first" >&2; exit 1; }
+
+# POST <url> <file> <label>: print HTTP status + a short slice of the body. A non-2xx/409 response is counted in FAILC for return value control
+post() {
+  local url="$1" file="$2" label="$3" code
+  code="$(curl -sS -o "$TMP/resp" -w '%{http_code}' -H 'Content-Type: application/json' -X POST --data-binary @"$file" "$url")"
+  case "$code" in
+    2*)  printf '  \033[1;32mOK  %s\033[0m (%s)\n' "$label" "$code" ;;
+    409) printf '  \033[1;33mEXISTS %s\033[0m (409 — leaving as-is)\n' "$label" ;;
+    *)   printf '  \033[1;31mFAIL %s\033[0m (%s): %s\n' "$label" "$code" "$(head -c 300 "$TMP/resp")"; FAILC=$((FAILC+1)) ;;
+  esac
+}
+
+echo "== 1/5 create project '$PROJECT' =="
+cat > "$TMP/project.json" <<JSON
+{
+  "id": "$PROJECT",
+  "name": "$PROJECT",
+  "description": "Manual SQL + Kafka source tests (Postgres 'patients' table + 'redcap-patients' topic).",
+  "schemaUrlPrefix": "https://aiccelerate.eu/fhir/StructureDefinition/",
+  "mappingUrlPrefix": "https://aiccelerate.eu/fhir/mappings/",
+  "schemas": [],
+  "mappings": []
+}
+JSON
+post "$BASE/projects" "$TMP/project.json" "project"
+
+echo "== 2/5 create Patient schema (Ext-patient StructureDefinition) =="
+post "$BASE/projects/$PROJECT/schemas?format=StructureDefinition" "$SCHEMA_FILE" "schema Ext-patient"
+
+echo "== 3/5 create patient mapping =="
+post "$BASE/projects/$PROJECT/mappings" "$MAPPING_FILE" "mapping patient-mapping"
+
+echo "== 4/5 create SQL job (Postgres 'patients' -> FHIR) =="
+cat > "$TMP/sql-job.json" <<'JSON'
+{
+  "id": "sql-patient-job",
+  "name": "sql-patient-job",
+  "sourceSettings": {
+    "sql": {
+      "jsonClass": "SqlSourceSettings",
+      "name": "sql",
+      "sourceUri": "https://ignifyr.io/test-flow/sql",
+      "databaseUrl": "jdbc:postgresql://postgres:5432/ignifyr",
+      "username": "ignifyr",
+      "password": "ignifyr"
+    }
+  },
+  "sinkSettings": {
+    "jsonClass": "FhirRepositorySinkSettings",
+    "fhirRepoUrl": "http://repofyr:8080/fhir"
+  },
+  "mappings": [
+    {
+      "name": "patient-mapping",
+      "mappingRef": "https://aiccelerate.eu/fhir/mappings/patient-mapping",
+      "sourceBinding": {
+        "source": {
+          "jsonClass": "SqlSource",
+          "tableName": "patients",
+          "sourceRef": "sql"
+        }
+      }
+    }
+  ],
+  "dataProcessingSettings": { "saveErroneousRecords": false, "archiveMode": "off" }
+}
+JSON
+post "$BASE/projects/$PROJECT/jobs" "$TMP/sql-job.json" "job sql-patient-job"
+
+echo "== 5/5 create Kafka job (redcap-patients topic -> FHIR) =="
+cat > "$TMP/kafka-job.json" <<'JSON'
+{
+  "id": "kafka-redcap-job",
+  "name": "kafka-redcap-job",
+  "sourceSettings": {
+    "kafka": {
+      "jsonClass": "KafkaSourceSettings",
+      "name": "kafka",
+      "sourceUri": "https://ignifyr.io/test-flow/redcap-kafka",
+      "bootstrapServers": "kafka:9092",
+      "asStream": true
+    }
+  },
+  "sinkSettings": {
+    "jsonClass": "FhirRepositorySinkSettings",
+    "fhirRepoUrl": "http://repofyr:8080/fhir"
+  },
+  "mappings": [
+    {
+      "name": "patient-mapping",
+      "mappingRef": "https://aiccelerate.eu/fhir/mappings/patient-mapping",
+      "sourceBinding": {
+        "source": {
+          "jsonClass": "KafkaSource",
+          "topicName": "redcap-patients",
+          "options": { "startingOffsets": "earliest" },
+          "sourceRef": "kafka"
+        }
+      }
+    }
+  ],
+  "dataProcessingSettings": { "saveErroneousRecords": false, "archiveMode": "off" }
+}
+JSON
+post "$BASE/projects/$PROJECT/jobs" "$TMP/kafka-job.json" "job kafka-redcap-job"
+
+echo
+if [ "$FAILC" -ne 0 ]; then
+  printf '\033[1;31m%s create(s) failed — see the FAIL lines above.\033[0m\n' "$FAILC"
+  exit 1
+fi
+
+# Seed the Kafka topic so 'kafka-redcap-job' has data
+echo "== seeding the 'redcap-patients' topic (reset + publish once) =="
+if MSYS_NO_PATHCONV=1 docker exec itf-kafka true 2>/dev/null; then
+  MSYS_NO_PATHCONV=1 docker exec itf-kafka /opt/kafka/bin/kafka-topics.sh \
+    --bootstrap-server localhost:9092 --delete --topic redcap-patients >/dev/null 2>&1 || true
+  sleep 5
+  if MSYS_NO_PATHCONV=1 docker exec -i itf-kafka /opt/kafka/bin/kafka-console-producer.sh \
+       --bootstrap-server localhost:9092 --topic redcap-patients < "$SCRIPT_DIR/data/redcap-patients.ndjson" 2>/dev/null; then
+    printf '  \033[1;32mOK  topic seeded with 3 records\033[0m\n'
+  else
+    printf '  \033[1;33mWARN could not publish to the topic - seed it manually before running the Kafka job\033[0m\n'
+  fi
+else
+  printf '  \033[1;33mWARN itf-kafka not running - start the stack, then seed the topic before the Kafka job\033[0m\n'
+fi
+
+echo
+echo "Done. Open the web UI -> project '$PROJECT' -> Jobs:"
+echo "  http://localhost/dt4h/ignifyr"
+echo "Run 'sql-patient-job' (batch) and 'kafka-redcap-job' (streaming) - both are ready to run."

test-flow/data/sql-source-seed.sql

@@ -0,0 +1,21 @@
+-- Seed data for the manual-flow SQL source (Postgres)
+CREATE TABLE patients
+(
+    pid                varchar(4) NOT NULL,
+    gender             varchar(8),
+    "birthDate"        date,
+    "deceasedDateTime" timestamp,
+    "homePostalCode"   varchar(8)
+);
+
+INSERT INTO patients (pid, gender, "birthDate", "deceasedDateTime", "homePostalCode") VALUES
+    ('p1',  'male',   DATE '2000-05-10', NULL,                          NULL),
+    ('p2',  'male',   DATE '1985-05-08', TIMESTAMP '2017-03-10 00:00:00','G02547'),
+    ('p3',  'male',   DATE '1997-02-01', NULL,                          NULL),
+    ('p4',  'male',   DATE '1999-06-05', NULL,                          'H10564'),
+    ('p5',  'male',   DATE '1965-10-01', TIMESTAMP '2019-04-21 00:00:00','G02547'),
+    ('p6',  'female', DATE '1991-03-01', NULL,                          NULL),
+    ('p7',  'female', DATE '1972-10-25', NULL,                          'V13135'),
+    ('p8',  'female', DATE '2010-01-10', NULL,                          'Z54564'),
+    ('p9',  'female', DATE '1999-05-12', NULL,                          NULL),
+    ('p10', 'female', DATE '2003-11-01', NULL,                          NULL);

test-flow/docker-compose.test.yml

@@ -63,6 +63,23 @@ services:
       - "9092:9092"
     networks: [itf-net]
 
+# For testing on web UI
+  postgres:
+    image: postgres:16
+    container_name: itf-postgres
+    environment:
+      - POSTGRES_DB=ignifyr
+      - POSTGRES_USER=ignifyr
+      - POSTGRES_PASSWORD=ignifyr
+    volumes:
+      - ./data/sql-source-seed.sql:/docker-entrypoint-initdb.d/seed.sql:ro
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U ignifyr -d ignifyr"]
+      interval: 10s
+      timeout: 5s
+      retries: 6
+    networks: [itf-net]
+
   ignifyr:
     image: srdc/ignifyr-server:latest
     container_name: itf-ignifyr

test-flow/run-manual-flow.sh

@@ -15,6 +15,7 @@
 #   streaming  folder-watch streaming (drop a CSV into a watched dir -> FHIR)
 #   scheduling cron-scheduled batch (fires every minute)
 #   kafka      REDCap-simulated Kafka streaming (publish records to a topic -> FHIR)
+#   sql        Postgres table -> FHIR batch mapping (SQL data source)
 #
 # Each mapped Patient carries identifier system = the job's sourceUri, so verification is an exact
 # FHIR search per behavior (no reliance on hashed ids). Jobs run through the engine CLI on the
@@ -34,7 +35,7 @@
 #   ./run-manual-flow.sh --no-dis        # with --with-web: don't seed the UI from data-ingestion-suite
 #   ./run-manual-flow.sh --with-efk      # also run Elasticsearch+Fluentd+Kibana (Executions dashboard)
 #   ./run-manual-flow.sh --skip-build    # reuse existing jars/image/web-dist
-#   ./run-manual-flow.sh --only batch    # run a single behavior (plugins|batch|archive|streaming|scheduling|kafka)
+#   ./run-manual-flow.sh --only batch    # run a single behavior (plugins|batch|archive|streaming|scheduling|kafka|sql)
 #   ./run-manual-flow.sh --down          # tear the stack down and exit
 set -euo pipefail
 
@@ -210,7 +211,7 @@ if [ "$WITH_EFK" = "1" ]; then
   "${COMPOSE[@]}" up -d elasticsearch fluentd kibana
 fi
 
-"${COMPOSE[@]}" up -d --wait mongo repofyr kafka ignifyr
+"${COMPOSE[@]}" up -d --wait mongo repofyr kafka postgres ignifyr
 
 # Wait for the FHIR server and the ignifyr REST server to answer.
 log "Waiting for repofyr and ignifyr-server"
@@ -228,10 +229,12 @@ fi
 
 # ----- helpers -----------------------------------------------------------------
 # Run a job file through the engine CLI on the enterprise classpath, isolated db/checkpoint.
+# The checkpoint and db for this job are wiped first
 run_job() {
   local name="$1" job="$2"; shift 2
   docker exec "$@" itf-ignifyr sh -c \
-    "java -Dconfig.file=$CONF_IN_IMAGE \
+    "rm -rf /workspace/clichk/$name /workspace/clidb/$name; \
+     java -Dconfig.file=$CONF_IN_IMAGE \
      -Dignifyr.mappings.repository.folder-path=/workspace/cli-mappings \
      -Dignifyr.mappings.schemas.repository.folder-path=/workspace/cli-schemas \
      -Dignifyr.db-path=/workspace/clidb/$name -Dspark.checkpoint-dir=/workspace/clichk/$name \
@@ -279,9 +282,9 @@ if want streaming; then
   rm -f "$SCRIPT_DIR/watch/patients/"*.csv 2>/dev/null || true
   # Start the streaming job (self-terminates after 100s via `timeout`), detached from this script.
   run_job streaming streaming-watch-job.json -d
-  sleep 25   # let the streaming query initialise before dropping the file
+  sleep 40   # let the streaming query initialise before dropping the file
   cp "$SCRIPT_DIR/data/stream-patients.csv" "$SCRIPT_DIR/watch/patients/stream-patients.csv"
-  if await_patient "https://ignifyr.io/test-flow/stream" "sp1" 90; then ok "streaming processed the dropped file (Patient sp1)"; else fail "streaming: Patient sp1 not found after drop"; fi
+  if await_patient "https://ignifyr.io/test-flow/stream" "sp1" 180; then ok "streaming processed the dropped file (Patient sp1)"; else fail "streaming: Patient sp1 not found after drop"; fi
   docker exec itf-ignifyr sh -c "pkill -f streaming-watch-job || true" >/dev/null 2>&1 || true
 fi
 
@@ -297,15 +300,25 @@ fi
 # ----- Kafka (REDCap-simulated) ------------------------------------------------
 if want kafka; then
   log "Kafka streaming (REDCap simulated via raw Kafka)"
-  # Publish REDCap-shaped records to the topic through the broker container's console producer.
+  MSYS_NO_PATHCONV=1 docker exec itf-kafka /opt/kafka/bin/kafka-topics.sh \
+    --bootstrap-server localhost:9092 --delete --topic redcap-patients >/dev/null 2>&1 || true
+  sleep 5
+  # Publish REDCap-shaped records to the (fresh) topic through the broker container's console producer.
   MSYS_NO_PATHCONV=1 docker exec -i itf-kafka /opt/kafka/bin/kafka-console-producer.sh \
     --bootstrap-server localhost:9092 --topic redcap-patients \
     < "$SCRIPT_DIR/data/redcap-patients.ndjson" || fail "could not publish to Kafka"
   run_job kafka kafka-redcap-job.json -d
-  if await_patient "https://ignifyr.io/test-flow/redcap-kafka" "rp1" 100; then ok "Kafka streaming consumed topic (Patient rp1)"; else fail "kafka: Patient rp1 not found"; fi
+  if await_patient "https://ignifyr.io/test-flow/redcap-kafka" "rp1" 180; then ok "Kafka streaming consumed topic (Patient rp1)"; else fail "kafka: Patient rp1 not found"; fi
   docker exec itf-ignifyr sh -c "pkill -f kafka-redcap-job || true" >/dev/null 2>&1 || true
 fi
 
+# ----- sql ----------------------------------------------------------
+if want sql; then
+  log "SQL (Postgres) -> FHIR"
+  run_job sql sql-patient-job.json
+  if await_patient "https://ignifyr.io/test-flow/sql" "p1" 60; then ok "sql read the Postgres table and produced Patient p1"; else fail "sql: Patient p1 not found in repofyr"; fi
+fi
+
 # ----- summary -----------------------------------------------------------------
 log "Summary"
 printf '  passed: %s   failed: %s\n' "$PASS" "$FAILC"
@@ -320,6 +333,7 @@ $EFK_LINE
   Ignifyr REST : $IGNIFYR         (e.g. curl $IGNIFYR/projects)
   Repofyr FHIR : $REPOFYR   (e.g. curl "$REPOFYR/Patient?_summary=count")
   Kafka broker : localhost:9092
+  Postgres     : host=postgres port=5432 db=ignifyr user=ignifyr pass=ignifyr (from the server; table 'patients')
   list-plugins : docker exec itf-ignifyr java -cp $JAR_IN_IMAGE io.ignifyr.engine.Boot list-plugins
 Tear down with:  $0 --down
 EOF