MongoDB Read Model vs PostgreSQL Reconcile Guard

id: mongodb-vs-postgresql-read-model-reconcile-guard
namespace: company.team

description: |
  Reconcile a denormalized MongoDB read model against its PostgreSQL source of truth. PostgreSQL is
  read first for the authoritative keyset and a per-row version, then the read-model documents are
  read from MongoDB. A document whose key is gone from PostgreSQL is ORPHANED. A document whose
  version is behind the row it mirrors is STALE. When auto_remediate is set the orphans are deleted
  and the stale documents refreshed from the database, each only behind an approval gate. Both sides
  are then read again and the run fails if any divergence remains.

inputs:
  - id: pg_host
    type: STRING
    displayName: PostgreSQL host
    defaults: "localhost"

  - id: pg_port
    type: INT
    displayName: PostgreSQL port
    defaults: 5432

  - id: pg_db
    type: STRING
    displayName: PostgreSQL database
    defaults: "postgres"

  - id: pg_user
    type: STRING
    displayName: PostgreSQL user
    defaults: "postgres"

  - id: pg_password
    type: STRING
    displayName: PostgreSQL password
    defaults: "{{ secret('POSTGRES_PASSWORD') }}"

  - id: pg_table
    type: STRING
    displayName: Source of truth table
    description: Table that holds the authoritative live rows. A deleted row is gone
      from it.
    defaults: "orders"

  - id: pg_key_column
    type: STRING
    displayName: Key column
    description: Primary key column that identifies a row and its read-model document.
    defaults: "order_id"

  - id: pg_version_column
    type: STRING
    displayName: Version column
    description: Column that increases on every update, compared against the
      document version.
    defaults: "version"

  - id: pg_payload_column
    type: STRING
    displayName: Payload column
    description: A denormalized column copied into the document when a stale
      document is refreshed.
    defaults: "status"

  - id: mongo_uri
    type: STRING
    displayName: MongoDB connection string
    defaults: "{{ secret('MONGODB_URI') }}"

  - id: mongo_db
    type: STRING
    displayName: MongoDB database
    defaults: "appdb"

  - id: mongo_collection
    type: STRING
    displayName: Read-model collection
    defaults: "orders_view"

  - id: doc_key_field
    type: STRING
    displayName: Document key field
    description: Field in the document that carries the row key. Its value matches
      pg_key_column.
    defaults: "order_id"

  - id: doc_version_field
    type: STRING
    displayName: Document version field
    defaults: "version"

  - id: doc_payload_field
    type: STRING
    displayName: Document payload field
    defaults: "status"

  - id: batch_limit
    type: INT
    displayName: Rows per run
    description: Upper bound on the keyset read from each side in one run, so a huge
      table cannot build an unbounded diff.
    defaults: 5000

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate, for any delete or
      refresh to run.
    defaults: true

variables:
  jdbc_url: "jdbc:postgresql://{{ inputs.pg_host }}:{{ inputs.pg_port }}/{{
    inputs.pg_db }}"
  diff_prog: >-
    .pg as $pg | .mg as $mg | .kf as $kf | .vf as $vf | ($pg | map({
    (.k|tostring): {v: .v, p: .p} }) | add // {}) as $pgm | ($mg | map({ k:
    .[$kf], v: .[$vf] })) as $mgx | ($mgx | map(.k|tostring)) as $mgkeys | ([
    $mgx[]
         | (.k|tostring) as $ks
         | if ($pgm[$ks] == null) then {code: "ORPHAN", key: .k, pg_version: null, mg_version: .v}
           elif (.v < $pgm[$ks].v) then {code: "STALE", key: .k, pg_version: $pgm[$ks].v, mg_version: .v, new_version: $pgm[$ks].v, new_payload: $pgm[$ks].p}
           else empty end ]) as $findings
    | ([ $pg[] | select(([.k|tostring] - $mgkeys) | length > 0) | .k ]) as
    $misses | {findings: $findings, misses: $misses}

tasks:
  - id: pg_read
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Read-only. The authoritative keyset, each row version and the
      payload column, in key order.
    url: "{{ render(vars.jdbc_url) }}"
    username: "{{ inputs.pg_user }}"
    password: "{{ inputs.pg_password }}"
    fetchType: FETCH
    sql: >-
      SELECT {{ inputs.pg_key_column }} AS k, {{ inputs.pg_version_column }} AS
      v, {{ inputs.pg_payload_column }} AS p FROM {{ inputs.pg_table }} ORDER BY
      {{ inputs.pg_key_column }} LIMIT {{ inputs.batch_limit }}

  - id: mongo_read
    type: io.kestra.plugin.mongodb.Find
    description: Read-only. Every read-model document with its key and version, _id dropped.
    connection:
      uri: "{{ inputs.mongo_uri }}"
    database: "{{ inputs.mongo_db }}"
    collection: "{{ inputs.mongo_collection }}"
    filter: {}
    projection:
      _id: 0
    limit: "{{ inputs.batch_limit }}"

  - id: diff
    type: io.kestra.plugin.core.output.OutputValues
    description: Diff both sides. Orphan when the document key is gone from
      PostgreSQL, stale when the document version is behind the row version.
    values:
      result: >-
        {{ {'pg': outputs.pg_read.rows, 'mg': outputs.mongo_read.rows, 'kf':
        inputs.doc_key_field, 'vf': inputs.doc_version_field} |
        jq(vars.diff_prog) | first | toJson }}

  - id: plan
    type: io.kestra.plugin.core.output.OutputValues
    description: The MongoDB delete filter for the orphan keys and the per-document
      refresh list for the stale keys.
    values:
      orphan_filter: >-
        {{ {'f': inputs.doc_key_field, 'rows':
        fromJson(outputs.diff.values.result).findings} | jq('.f as $f | {($f):
        {"$in": [.rows[] | select(.code=="ORPHAN") | .key]}}') | first | toJson
        }}
      stale: >-
        {{ {'f': inputs.doc_key_field, 'vf': inputs.doc_version_field, 'pf':
        inputs.doc_payload_field, 'rows':
        fromJson(outputs.diff.values.result).findings} | jq('.f as $f | .vf as
        $vf | .pf as $pf | [.rows[] | select(.code=="STALE") | {filter_str:
        ({($f): .key} | tojson), set_str: ({"$set": {($vf): .new_version, ($pf):
        .new_payload}} | tojson)}]') | first | toJson }}

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Reconcile of {{ inputs.mongo_db }}.{{ inputs.mongo_collection }} against
      {{ render(vars.jdbc_url) }}: {{
      fromJson(outputs.diff.values.result).findings | length }} divergences ({{
      fromJson(outputs.diff.values.result).findings | jq('[.[] |
      select(.code=="ORPHAN")] | length') | first }} orphan, {{
      fromJson(outputs.diff.values.result).findings | jq('[.[] |
      select(.code=="STALE")] | length') | first }} stale), {{
      fromJson(outputs.diff.values.result).misses | length }} rows not yet in
      the read model. {{ fromJson(outputs.diff.values.result).findings |
      jq('map("\(.code) \(.key) db=\(.pg_version // "gone") doc=\(.mg_version)")
      | join("; ")') | first }}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(outputs.diff.values.result).findings | length > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'MongoDB vs PostgreSQL reconcile on ' ~ inputs.mongo_db ~ '.' ~ inputs.mongo_collection ~ ': ' ~ (fromJson(outputs.diff.values.result).findings | jq('map("\(.code) \(.key)") | join(", ")') | first) ~ '.' ~ (inputs.auto_remediate and not inputs.dry_run ? ' Remediation armed, waiting for approval.' : ' No change applied (auto_remediate=' ~ inputs.auto_remediate ~ ', dry_run=' ~ inputs.dry_run ~ ').')} | toJson }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: approval
            type: io.kestra.plugin.core.flow.Pause
            description: Hold before any destructive write. Resume with approved true to
              apply, anything else leaves the read model untouched.
            onResume:
              - id: approved
                type: BOOL
                required: false
                description: Set true to delete the orphans and refresh the stale documents.

          - id: apply_gate
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approval.onResume.approved ?? false }}"
            then:
              - id: evict_gate
                type: io.kestra.plugin.core.flow.If
                condition: "{{ (fromJson(outputs.diff.values.result).findings | jq('[.[] |
                  select(.code==\"ORPHAN\")] | length') | first) > 0 }}"
                then:
                  - id: delete_orphans
                    type: io.kestra.plugin.mongodb.Delete
                    description: Delete the documents whose row no longer exists in the source of
                      truth.
                    connection:
                      uri: "{{ inputs.mongo_uri }}"
                    database: "{{ inputs.mongo_db }}"
                    collection: "{{ inputs.mongo_collection }}"
                    operation: DELETE_MANY
                    filter: "{{ outputs.plan.values.orphan_filter }}"

              - id: stale_gate
                type: io.kestra.plugin.core.flow.If
                condition: "{{ (fromJson(outputs.diff.values.result).findings | jq('[.[] |
                  select(.code==\"STALE\")] | length') | first) > 0 }}"
                then:
                  - id: remediate_stale
                    type: io.kestra.plugin.core.flow.Loop
                    description: Refresh each stale document from the current database row. Serial
                      so the writes do not race.
                    concurrencyLimit: 1
                    values: "{{ outputs.plan.values.stale }}"
                    tasks:
                      - id: refresh_doc
                        type: io.kestra.plugin.mongodb.Update
                        connection:
                          uri: "{{ inputs.mongo_uri }}"
                        database: "{{ inputs.mongo_db }}"
                        collection: "{{ inputs.mongo_collection }}"
                        operation: UPDATE_ONE
                        filter: "{{ fromJson(item.value).filter_str }}"
                        document: "{{ fromJson(item.value).set_str }}"

              - id: reverify_pg
                type: io.kestra.plugin.jdbc.postgresql.Query
                description: Read-only. Read the source of truth again after the writes.
                url: "{{ render(vars.jdbc_url) }}"
                username: "{{ inputs.pg_user }}"
                password: "{{ inputs.pg_password }}"
                fetchType: FETCH
                sql: >-
                  SELECT {{ inputs.pg_key_column }} AS k, {{
                  inputs.pg_version_column }} AS v, {{ inputs.pg_payload_column
                  }} AS p FROM {{ inputs.pg_table }} ORDER BY {{
                  inputs.pg_key_column }} LIMIT {{ inputs.batch_limit }}

              - id: reverify_mongo
                type: io.kestra.plugin.mongodb.Find
                description: Read the read-model documents again, so the proof comes from
                  MongoDB itself.
                connection:
                  uri: "{{ inputs.mongo_uri }}"
                database: "{{ inputs.mongo_db }}"
                collection: "{{ inputs.mongo_collection }}"
                filter: {}
                projection:
                  _id: 0
                limit: "{{ inputs.batch_limit }}"

              - id: reverify_diff
                type: io.kestra.plugin.core.output.OutputValues
                description: Re-run the same diff. An orphan still present or a document still
                  behind is unresolved.
                values:
                  remaining: >-
                    {{ {'pg': outputs.reverify_pg.rows, 'mg':
                    outputs.reverify_mongo.rows, 'kf': inputs.doc_key_field,
                    'vf': inputs.doc_version_field} | jq(vars.diff_prog) | first
                    | toJson }}

              - id: verify_result
                type: io.kestra.plugin.core.flow.If
                condition: "{{ fromJson(outputs.reverify_diff.values.remaining).findings |
                  length > 0 }}"
                then:
                  - id: remediation_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      Reconcile did not converge: {{
                      fromJson(outputs.reverify_diff.values.remaining).findings
                      | jq('map("\(.code) \(.key)") | join(", ")') | first }}
                      still diverge. A row may have changed during the run or
                      the connecting MongoDB user may not delete or update these
                      documents.
                else:
                  - id: remediated
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Reconciled and verified: deleted the orphan documents,
                      refreshed the stale documents, 0 divergence left.
            else:
              - id: withheld
                type: io.kestra.plugin.core.log.Log
                message: "Approval was not granted, so no document was deleted or refreshed. The
                  divergence stands."
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "The read model matches the source of truth: every document mirrors a
          live row at its current version and no deleted row is still a
          document."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'MongoDB vs PostgreSQL reconcile guard failed on ' ~ inputs.mongo_db ~ '.' ~ inputs.mongo_collection ~ ' (execution ' ~ execution.id ~ '). Check the logs.'} | toJson }}

triggers:
  - id: hourly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled and always in safe mode.
    cron: "25 * * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true