Reconcile the bank account and close the month, with an evidence pack

id: month-end-bank-reconciliation-close
namespace: company.team

description: |
  Month-end cash close: match the bank statement against the general ledger in five passes,
  classify every leftover item, build the reconciliation statement, propose journals, and
  lock the period only when the difference is zero and nothing is unexplained.

inputs:
  - id: period
    type: STRING
    displayName: Period (YYYY-MM)
    description: The month to reconcile and close.
    defaults: "2026-09"
    validator: ^\d{4}-(0[1-9]|1[0-2])$

  - id: cash_account
    type: STRING
    displayName: Cash account
    description: The general ledger account that mirrors the bank account.
    defaults: "1010"

  - id: scenario
    type: SELECT
    displayName: Demo scenario
    description: >
      CLEAN: every difference is explained by timing or by bank charges the
      ledger has not booked yet. ISSUES: a duplicate ledger posting, a
      transposed check amount and an unidentified deposit, which must block the
      close.
    values:
      - CLEAN
      - ISSUES
    defaults: CLEAN

  - id: reference_window_days
    type: INT
    displayName: Reference match window (days)
    description: How far apart dates may be when the reference also matches. Checks
      often clear one to three weeks after they are written.
    defaults: 45

  - id: date_tolerance_days
    type: INT
    displayName: Amount-only match window (days)
    description: How far apart dates may be when only the amount matches. Keep it
      short, because equal amounts are common.
    defaults: 5

  - id: transit_window_days
    type: INT
    displayName: Deposit in transit window (days)
    description: A ledger receipt from the last days of the period that the bank has
      not shown yet is a deposit in transit. Older ones are unexplained.
    defaults: 3

  - id: auto_post_limit
    type: FLOAT
    displayName: Auto-post limit
    description: Proposed journals whose total absolute value is under this are
      posted without approval. Above it, the run needs approved_by.
    defaults: 500.0

  - id: approved_by
    type: STRING
    displayName: Approved by
    description: Name of the controller who reviewed the proposed journals. Needed
      when they exceed auto_post_limit.
    defaults: ""

variables:
  close_key: "cash_close_{{ inputs.cash_account }}"
  pack_dir: "close/{{ inputs.cash_account }}/{{ inputs.period }}"

concurrency:
  limit: 1

triggers:
  - id: second_business_day
    type: io.kestra.plugin.core.trigger.Schedule
    description: Run on the 2nd of each month for the month before, once the
      statement has arrived. Disabled until your data sources are wired in.
    cron: "0 6 2 * *"
    disabled: true

tasks:
  # ---------------------------------------------------------------------------
  # 1. Period control: never close out of order, never close twice
  # ---------------------------------------------------------------------------
  - id: period_control
    type: io.kestra.plugin.core.output.OutputValues
    description: The last closed period for this account, and the period that has to
      come before this one.
    values:
      last_closed: "{{ (kv(render(vars.close_key), errorOnMissing=false) ?? {'period':
        ''}) | jq('.period') | first }}"
      previous_period: "{{ (inputs.period ~ '-01') | date('yyyy-MM-dd') | dateAdd(-1,
        'MONTHS') | date('yyyy-MM') }}"
      period_end: "{{ (inputs.period ~ '-01') | date('yyyy-MM-dd') | dateAdd(1,
        'MONTHS') | dateAdd(-1, 'DAYS') | date('yyyy-MM-dd') }}"

  - id: period_gate
    type: io.kestra.plugin.core.flow.If
    description: Refuse a period that is already closed, or one whose previous
      period is still open.
    condition: >-
      {{ outputs.period_control.values.last_closed != ''
         and (outputs.period_control.values.last_closed >= inputs.period
              or outputs.period_control.values.last_closed != outputs.period_control.values.previous_period) }}
    then:
      - id: period_refused
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: >-
          {{ outputs.period_control.values.last_closed >= inputs.period
             ? 'Period ' ~ inputs.period ~ ' is already closed (last closed: ' ~ outputs.period_control.values.last_closed ~ '). Reopen it deliberately by deleting KV ' ~ render(vars.close_key) ~ '.'
             : 'Period ' ~ outputs.period_control.values.previous_period ~ ' must be closed before ' ~ inputs.period ~ ' (last closed: ' ~ outputs.period_control.values.last_closed ~ ').' }}

  # ---------------------------------------------------------------------------
  # 2. Match, classify, reconcile
  # ---------------------------------------------------------------------------
  - id: reconcile
    type: io.kestra.plugin.jdbc.duckdb.Queries
    description: >
      Load the ledger and the bank statement, run five matching passes, classify
      every item left over, compute the reconciliation statement and the
      proposed journals, and prove the arithmetic: the unexplained difference
      must equal the sum of the unexplained items.
    outputFiles:
      - matches
      - exceptions
      - journals
      - statement
    fetchType: FETCH_ONE
    sql: |
      SELECT setseed(0.2026);
      SET VARIABLE period_start = DATE '{{ inputs.period }}-01';
      SET VARIABLE period_end = DATE '{{ outputs.period_control.values.period_end }}';
      SET VARIABLE opening = 250000.00::DECIMAL(14, 2);

      -- =======================================================================
      -- Demo data. Replace these two tables with your ERP export and your bank
      -- statement (camt.053, MT940, BAI2 or CSV), keeping the same columns.
      -- =======================================================================
      CREATE TABLE gl_source AS
      WITH wires AS (
        SELECT 'INV-' || (10000 + i) AS ref, 'Customer receipt INV-' || (10000 + i) AS memo,
               getvariable('period_start') + ((i * 7) % 30)::INT AS posted_on,
               (500 + (i * 7919) % 9500 + ((i * 37) % 100) / 100.0)::DECIMAL(14, 2) AS amount, 'WIRE' AS kind
        FROM range(1, 121) t(i)
      ), card AS (
        SELECT 'CARD-' || strftime(getvariable('period_start') + d::INT, '%Y%m%d') AS ref,
               'Card sales ' || strftime(getvariable('period_start') + d::INT, '%Y-%m-%d') AS memo,
               getvariable('period_start') + d::INT AS posted_on,
               (1200 + (d * 137) % 800 + (d % 7) * 0.35)::DECIMAL(14, 2) AS amount, 'CARD' AS kind
        FROM range(0, 30) t(d)
      ), checks AS (
        SELECT 'CHK ' || (5000 + k) AS ref, 'Vendor check ' || (5000 + k) AS memo,
               getvariable('period_start') + ((k * 11) % 30)::INT AS posted_on,
               CASE WHEN k = 12 THEN -1245.30 ELSE -(200 + (k * 911) % 4000 + ((k * 13) % 100) / 100.0) END::DECIMAL(14, 2) AS amount, 'CHECK' AS kind
        FROM range(1, 61) t(k)
      ), payroll AS (
        SELECT 'PAYROLL-' || n AS ref, 'Payroll run ' || n AS memo,
               CASE n WHEN 1 THEN getvariable('period_start') + 14 ELSE getvariable('period_end') END AS posted_on,
               CASE n WHEN 1 THEN -48250.00 ELSE -48410.00 END::DECIMAL(14, 2) AS amount, 'ACH' AS kind
        FROM range(1, 3) t(n)
      )
      SELECT * FROM wires UNION ALL SELECT * FROM card UNION ALL SELECT * FROM checks UNION ALL SELECT * FROM payroll
      {% if inputs.scenario == 'ISSUES' %}
      -- A receipt posted twice by mistake.
      UNION ALL SELECT ref, memo || ' (posted again)', posted_on, amount, kind FROM wires WHERE ref = 'INV-10037'
      {% endif %};

      CREATE TABLE gl AS
      SELECT row_number() OVER (ORDER BY posted_on, ref, memo) AS gl_id, * FROM gl_source
      WHERE posted_on BETWEEN getvariable('period_start') AND getvariable('period_end');

      -- The processor pays card sales out net of a 2.9% fee, two business days later, so
      -- weekend sales arrive together in one payout.
      CREATE TABLE card_schedule AS
      SELECT gl_id, posted_on,
             CASE dayofweek(posted_on + 2) WHEN 6 THEN posted_on + 4 WHEN 0 THEN posted_on + 3 ELSE posted_on + 2 END AS payout_on
      FROM gl WHERE kind = 'CARD';

      CREATE TABLE bank_source AS
      SELECT 'WIRE IN ' || replace(g.ref, '-', '') AS bank_ref, 'INCOMING WIRE CUSTOMER ' || replace(g.ref, '-', '') AS description,
             g.posted_on + (g.gl_id % 2)::INT AS value_date, g.amount
      FROM gl g WHERE g.kind = 'WIRE' AND g.memo NOT LIKE '%posted again%'
      UNION ALL
      SELECT 'PAYOUT ' || strftime(s.payout_on, '%Y%m%d'), 'CARD PROCESSOR PAYOUT ' || strftime(s.payout_on, '%Y-%m-%d'),
             s.payout_on, round(sum(g.amount) * 0.971, 2)::DECIMAL(14, 2)
      FROM card_schedule s JOIN gl g USING (gl_id) GROUP BY s.payout_on
      UNION ALL
      SELECT regexp_replace(g.ref, '[^0-9]', '', 'g'), 'CHECK PAID ' || regexp_replace(g.ref, '[^0-9]', '', 'g'),
             g.posted_on + (3 + g.gl_id % 9)::INT,
             {% if inputs.scenario == 'ISSUES' %}CASE WHEN g.ref = 'CHK 5012' THEN -1254.30 ELSE g.amount END{% else %}g.amount{% endif %}
      FROM gl g WHERE g.kind = 'CHECK'
      UNION ALL
      SELECT 'ACH PAYROLL', 'PAYROLL ACH DEBIT', g.posted_on, g.amount FROM gl g WHERE g.kind = 'ACH'
      UNION ALL SELECT 'FEE 0930', 'MONTHLY ACCOUNT SERVICE FEE', getvariable('period_end'), -45.00
      UNION ALL SELECT 'FEE WIRE', 'INCOMING WIRE FEES 12 X 3.00', getvariable('period_end'), -36.00
      UNION ALL SELECT 'INT 0930', 'INTEREST CREDIT', getvariable('period_end'), 312.48
      UNION ALL SELECT 'RTN 10055', 'RETURNED ITEM NSF ' || 'INV10055', getvariable('period_start') + 20,
                       -(SELECT amount FROM gl WHERE ref = 'INV-10055')
      {% if inputs.scenario == 'ISSUES' %}
      UNION ALL SELECT 'DEP 77812', 'DEPOSIT REFERENCE 77812', getvariable('period_start') + 17, 2500.00
      {% endif %};

      -- The statement only holds what the bank booked inside the period.
      CREATE TABLE bank AS
      SELECT row_number() OVER (ORDER BY value_date, bank_ref) AS bank_id, *
      FROM bank_source WHERE value_date BETWEEN getvariable('period_start') AND getvariable('period_end');

      -- =======================================================================
      -- Matching. Each pass only sees what earlier passes left. A pair is kept
      -- only when each side is the other's best candidate, so no line is used twice.
      -- =======================================================================
      CREATE MACRO norm_ref(r) AS regexp_replace(upper(r), '[^0-9]', '', 'g');
      CREATE TABLE matches (pass VARCHAR, bank_id BIGINT, gl_id BIGINT, bank_amount DECIMAL(14, 2), gl_amount DECIMAL(14, 2), note VARCHAR);

      -- Pass 1: same amount, same reference digits, within the reference window.
      INSERT INTO matches
      WITH c AS (
        SELECT b.bank_id, g.gl_id, b.amount AS ba, g.amount AS ga, abs(b.value_date - g.posted_on) AS dd
        FROM bank b JOIN gl g ON b.amount = g.amount AND norm_ref(b.bank_ref) = norm_ref(g.ref) AND norm_ref(g.ref) <> ''
        WHERE abs(b.value_date - g.posted_on) <= {{ inputs.reference_window_days }}
      ), r AS (SELECT *, row_number() OVER (PARTITION BY bank_id ORDER BY dd, gl_id) rb, row_number() OVER (PARTITION BY gl_id ORDER BY dd, bank_id) rg FROM c)
      SELECT '1 reference + amount', bank_id, gl_id, ba, ga, dd || ' day(s) apart' FROM r WHERE rb = 1 AND rg = 1;

      -- Pass 2: same amount within the date tolerance, no usable reference.
      INSERT INTO matches
      WITH c AS (
        SELECT b.bank_id, g.gl_id, b.amount AS ba, g.amount AS ga, abs(b.value_date - g.posted_on) AS dd
        FROM bank b JOIN gl g ON b.amount = g.amount
        WHERE abs(b.value_date - g.posted_on) <= {{ inputs.date_tolerance_days }}
          AND b.bank_id NOT IN (SELECT bank_id FROM matches) AND g.gl_id NOT IN (SELECT gl_id FROM matches)
      ), r AS (SELECT *, row_number() OVER (PARTITION BY bank_id ORDER BY dd, gl_id) rb, row_number() OVER (PARTITION BY gl_id ORDER BY dd, bank_id) rg FROM c)
      SELECT '2 amount + date', bank_id, gl_id, ba, ga, dd || ' day(s) apart' FROM r WHERE rb = 1 AND rg = 1;

      -- Pass 3: one payout against several card-sales postings, net of the processor fee.
      CREATE TABLE payout_groups AS
      SELECT b.bank_id, b.amount AS net, list(s.gl_id ORDER BY s.gl_id) AS gl_ids, sum(g.amount) AS gross
      FROM bank b
      JOIN card_schedule s ON s.payout_on = b.value_date
      JOIN gl g ON g.gl_id = s.gl_id
      WHERE b.description LIKE 'CARD PROCESSOR PAYOUT%' AND b.bank_id NOT IN (SELECT bank_id FROM matches)
      GROUP BY b.bank_id, b.amount
      HAVING b.amount / sum(g.amount) BETWEEN 0.95 AND 0.99;
      INSERT INTO matches
      SELECT '3 payout net of fees', p.bank_id, unnest(p.gl_ids), p.net, NULL,
             len(p.gl_ids) || ' sales, gross ' || p.gross || ', fee ' || (p.gross - p.net) FROM payout_groups p;

      -- Pass 4: same reference digits but a different amount whose difference is a
      -- multiple of 9, the signature of two swapped digits. Flagged, never auto-matched.
      CREATE TABLE suspected AS
      SELECT b.bank_id, g.gl_id, b.amount AS ba, g.amount AS ga, b.amount - g.amount AS diff
      FROM bank b JOIN gl g ON norm_ref(b.bank_ref) = norm_ref(g.ref) AND norm_ref(g.ref) <> ''
      WHERE b.amount <> g.amount AND (abs(b.amount - g.amount) * 100) % 900 = 0
        AND b.bank_id NOT IN (SELECT bank_id FROM matches) AND g.gl_id NOT IN (SELECT gl_id FROM matches);

      -- =======================================================================
      -- Classify everything left over.
      -- =======================================================================
      CREATE TABLE exceptions AS
      SELECT 'BANK' AS side, b.bank_id AS id, b.value_date AS on_date, b.bank_ref AS ref, b.description, b.amount,
             CASE
               WHEN b.bank_id IN (SELECT bank_id FROM suspected) THEN 'SUSPECTED_TRANSPOSITION'
               WHEN b.description LIKE '%SERVICE FEE%' OR b.description LIKE '%WIRE FEES%' THEN 'BANK_FEE'
               WHEN b.description LIKE 'INTEREST%' THEN 'INTEREST'
               WHEN b.description LIKE 'RETURNED ITEM%' THEN 'RETURNED_ITEM'
               ELSE 'UNIDENTIFIED_RECEIPT'
             END AS class
      FROM bank b WHERE b.bank_id NOT IN (SELECT bank_id FROM matches)
      UNION ALL
      SELECT 'BOOK', g.gl_id, g.posted_on, g.ref, g.memo, g.amount,
             CASE
               WHEN g.gl_id IN (SELECT gl_id FROM suspected) THEN 'SUSPECTED_TRANSPOSITION'
               WHEN g.kind = 'CHECK' THEN 'OUTSTANDING_CHECK'
               WHEN g.kind = 'CARD' AND g.gl_id IN (SELECT gl_id FROM card_schedule WHERE payout_on > getvariable('period_end')) THEN 'DEPOSIT_IN_TRANSIT'
               WHEN g.amount > 0 AND g.posted_on > getvariable('period_end') - {{ inputs.transit_window_days }} THEN 'DEPOSIT_IN_TRANSIT'
               WHEN g.memo LIKE '%posted again%' OR g.ref IN (SELECT ref FROM gl GROUP BY ref HAVING count(*) > 1) THEN 'DUPLICATE_POSTING'
               ELSE 'UNEXPLAINED_BOOK'
             END
      FROM gl g WHERE g.gl_id NOT IN (SELECT gl_id FROM matches);

      -- =======================================================================
      -- Proposed journals for what the bank knows and the ledger does not yet.
      -- =======================================================================
      CREATE TABLE journals AS
      SELECT row_number() OVER (ORDER BY account, memo) AS line, * FROM (
        SELECT '6810' AS account, 'Bank charges' AS account_name, -amount AS debit, 0.00 AS credit, description AS memo FROM exceptions WHERE class = 'BANK_FEE'
        UNION ALL SELECT '{{ inputs.cash_account }}', 'Cash', 0.00, -amount, description FROM exceptions WHERE class = 'BANK_FEE'
        UNION ALL SELECT '{{ inputs.cash_account }}', 'Cash', amount, 0.00, description FROM exceptions WHERE class = 'INTEREST'
        UNION ALL SELECT '7100', 'Interest income', 0.00, amount, description FROM exceptions WHERE class = 'INTEREST'
        UNION ALL SELECT '1200', 'Accounts receivable', -amount, 0.00, description FROM exceptions WHERE class = 'RETURNED_ITEM'
        UNION ALL SELECT '{{ inputs.cash_account }}', 'Cash', 0.00, -amount, description FROM exceptions WHERE class = 'RETURNED_ITEM'
        UNION ALL SELECT '6820', 'Card processing fees', sum(gross - net), 0.00, 'Processor fees on ' || count(*) || ' payouts' FROM payout_groups HAVING count(*) > 0
        UNION ALL SELECT '{{ inputs.cash_account }}', 'Cash', 0.00, sum(gross - net), 'Processor fees on ' || count(*) || ' payouts' FROM payout_groups HAVING count(*) > 0
      );

      -- =======================================================================
      -- Reconciliation statement.
      -- =======================================================================
      CREATE TABLE statement AS
      WITH v AS (
        SELECT getvariable('opening') + (SELECT coalesce(sum(amount), 0) FROM bank) AS bank_closing,
               getvariable('opening') + (SELECT coalesce(sum(amount), 0) FROM gl) AS book_closing,
               (SELECT coalesce(sum(amount), 0) FROM exceptions WHERE side = 'BOOK' AND class = 'DEPOSIT_IN_TRANSIT') AS in_transit,
               (SELECT coalesce(sum(amount), 0) FROM exceptions WHERE side = 'BOOK' AND class = 'OUTSTANDING_CHECK') AS outstanding,
               (SELECT coalesce(sum(debit - credit), 0) FROM journals WHERE account = '{{ inputs.cash_account }}') AS journal_effect,
               (SELECT coalesce(sum(CASE side WHEN 'BANK' THEN amount ELSE -amount END), 0)
                  FROM exceptions WHERE class IN ('UNIDENTIFIED_RECEIPT', 'UNEXPLAINED_BOOK', 'DUPLICATE_POSTING', 'SUSPECTED_TRANSPOSITION')) AS unexplained_sum
      )
      SELECT *, bank_closing + in_transit + outstanding AS adjusted_bank,
             book_closing + journal_effect AS adjusted_book,
             (bank_closing + in_transit + outstanding) - (book_closing + journal_effect) AS difference
      FROM v;

      COPY (SELECT m.*, b.value_date, b.bank_ref, g.posted_on, g.ref AS gl_ref FROM matches m LEFT JOIN bank b USING (bank_id) LEFT JOIN gl g USING (gl_id) ORDER BY pass, bank_id) TO '{{ outputFiles.matches }}' (HEADER, DELIMITER ',');
      COPY (SELECT * FROM exceptions ORDER BY class, on_date) TO '{{ outputFiles.exceptions }}' (HEADER, DELIMITER ',');
      COPY journals TO '{{ outputFiles.journals }}' (HEADER, DELIMITER ',');
      COPY statement TO '{{ outputFiles.statement }}' (HEADER, DELIMITER ',');

      SELECT
        (SELECT count(*) FROM bank) AS bank_lines,
        (SELECT count(*) FROM gl) AS gl_lines,
        (SELECT count(DISTINCT bank_id) FROM matches WHERE pass LIKE '1%') AS pass1,
        (SELECT count(DISTINCT bank_id) FROM matches WHERE pass LIKE '2%') AS pass2,
        (SELECT count(DISTINCT bank_id) FROM matches WHERE pass LIKE '3%') AS pass3_payouts,
        (SELECT count(*) FROM matches WHERE pass LIKE '3%') AS pass3_sales,
        (SELECT count(*) FROM suspected) AS suspected_transpositions,
        (SELECT count(*) FROM exceptions WHERE class = 'OUTSTANDING_CHECK') AS outstanding_checks,
        (SELECT count(*) FROM exceptions WHERE class = 'DEPOSIT_IN_TRANSIT') AS deposits_in_transit,
        (SELECT count(*) FROM exceptions WHERE class IN ('BANK_FEE', 'INTEREST', 'RETURNED_ITEM')) AS bank_items_to_book,
        (SELECT count(*) FROM exceptions WHERE class IN ('UNIDENTIFIED_RECEIPT', 'UNEXPLAINED_BOOK', 'DUPLICATE_POSTING', 'SUSPECTED_TRANSPOSITION')) AS unexplained,
        (SELECT coalesce(string_agg(class || ' ' || side || ' ' || ref || ' ' || amount, '; ' ORDER BY class, ref), '')
           FROM exceptions WHERE class IN ('UNIDENTIFIED_RECEIPT', 'UNEXPLAINED_BOOK', 'DUPLICATE_POSTING', 'SUSPECTED_TRANSPOSITION')) AS unexplained_items,
        (SELECT coalesce(string_agg('CHK ' || s.ga || ' booked vs ' || s.ba || ' paid (diff ' || s.diff || ')', '; '), '') FROM suspected s) AS transposition_detail,
        (SELECT round(sum(debit), 2) FROM journals) AS journal_total,
        (SELECT count(*) FROM journals) AS journal_lines,
        bank_closing, book_closing, in_transit, outstanding, journal_effect, adjusted_bank, adjusted_book, difference, unexplained_sum,
        difference = unexplained_sum AS arithmetic_proven
      FROM statement;

  - id: result
    type: io.kestra.plugin.core.output.OutputValues
    values:
      r: "{{ (outputs.reconcile.outputs | last).row | toJson }}"

  - id: log_reconciliation
    type: io.kestra.plugin.core.log.Log
    message: |
      Cash account {{ inputs.cash_account }}, period {{ inputs.period }} ({{ inputs.scenario }}).
      {{ outputs.result.values.r | jq('"Bank lines \(.bank_lines), ledger lines \(.gl_lines). Matched: pass 1 \(.pass1), pass 2 \(.pass2), pass 3 \(.pass3_payouts) payouts covering \(.pass3_sales) sales."') | first }}
      {{ outputs.result.values.r | jq('"Timing: \(.outstanding_checks) outstanding checks (\(.outstanding)), \(.deposits_in_transit) deposits in transit (\(.in_transit)). Bank items to book: \(.bank_items_to_book) in \(.journal_lines) journal lines."') | first }}
      {{ outputs.result.values.r | jq('"Bank closing \(.bank_closing), adjusted \(.adjusted_bank). Book closing \(.book_closing), adjusted \(.adjusted_book). Difference \(.difference)."') | first }}
      {{ outputs.result.values.r | jq('"Unexplained: \(.unexplained) item(s) summing to \(.unexplained_sum); arithmetic proven: \(.arithmetic_proven). \(.unexplained_items) \(.transposition_detail)"') | first }}

  - id: evidence
    type: io.kestra.plugin.core.namespace.UploadFiles
    description: The evidence pack, kept for every run, closed or not.
    namespace: "{{ flow.namespace }}"
    filesMap:
      "{{ render(vars.pack_dir) }}/matches.csv": "{{ outputs.reconcile.outputFiles.matches }}"
      "{{ render(vars.pack_dir) }}/exceptions.csv": "{{ outputs.reconcile.outputFiles.exceptions }}"
      "{{ render(vars.pack_dir) }}/proposed-journals.csv": "{{ outputs.reconcile.outputFiles.journals }}"
      "{{ render(vars.pack_dir) }}/statement.csv": "{{ outputs.reconcile.outputFiles.statement }}"

  # ---------------------------------------------------------------------------
  # 3. Decide: close, wait for approval, or block on exceptions
  # ---------------------------------------------------------------------------
  - id: verdict
    type: io.kestra.plugin.core.output.OutputValues
    values:
      state: >-
        {%- set r = outputs.result.values.r -%} {%- if not (r |
        jq('.arithmetic_proven') | first) -%}ARITHMETIC_ERROR {%- elseif (r |
        jq('.unexplained') | first) > 0 or (r | jq('.difference') | first) != 0
        -%}EXCEPTIONS {%- elseif (r | jq('.journal_total') | first ?? 0) >
        inputs.auto_post_limit and inputs.approved_by == '' -%}NEEDS_APPROVAL
        {%- else -%}READY{%- endif -%}

  - id: decide
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ outputs.verdict.values.state }}"
    cases:
      READY:
        - id: post_journals
          type: io.kestra.plugin.jdbc.duckdb.Queries
          description: Mark the proposed journals as posted, with who approved them. In
            real use, post them to the ERP here, for example through its API.
          inputFiles:
            journals.csv: "{{ outputs.reconcile.outputFiles.journals }}"
          outputFiles:
            - posted
          fetchType: NONE
          sql: |
            COPY (SELECT *, '{{ inputs.approved_by == '' ? 'auto (under ' ~ inputs.auto_post_limit ~ ')' : inputs.approved_by }}' AS approved_by,
                         '{{ execution.id }}' AS execution_id FROM read_csv('journals.csv'))
            TO '{{ outputFiles.posted }}' (HEADER, DELIMITER ',');

        - id: store_posted
          type: io.kestra.plugin.core.namespace.UploadFiles
          namespace: "{{ flow.namespace }}"
          filesMap:
            "{{ render(vars.pack_dir) }}/posted-journals.csv": "{{ outputs.post_journals.outputFiles.posted }}"

        - id: lock_period
          type: io.kestra.plugin.core.kv.Set
          description: The period lock. The next run refuses this period and accepts only
            the following one.
          key: "{{ render(vars.close_key) }}"
          kvType: JSON
          value: |
            {"period": "{{ inputs.period }}", "closed_at": "{{ now() }}", "execution_id": "{{ execution.id }}",
             "adjusted_balance": {{ outputs.result.values.r | jq('.adjusted_bank') | first }},
             "journal_total": {{ outputs.result.values.r | jq('.journal_total') | first }},
             "approved_by": "{{ inputs.approved_by == '' ? 'auto' : inputs.approved_by }}"}

        - id: log_closed
          type: io.kestra.plugin.core.log.Log
          message: "Period {{ inputs.period }} closed for cash account {{
            inputs.cash_account }}. Adjusted balance {{ outputs.result.values.r
            | jq('.adjusted_bank') | first }}. Evidence pack in {{
            render(vars.pack_dir) }}."

      NEEDS_APPROVAL:
        - id: approval_required
          type: io.kestra.plugin.core.execution.Fail
          description: The reconciliation is clean but the journals are above the
            auto-post limit. Nothing is posted or locked.
          errorMessage: >-
            Reconciliation is clean, but the proposed journals total {{
            outputs.result.values.r | jq('.journal_total') | first }}, above the
            auto-post limit of {{ inputs.auto_post_limit }}. Review {{
            render(vars.pack_dir) }}/proposed-journals.csv, then run again with
            approved_by set.

      ARITHMETIC_ERROR:
        - id: arithmetic_error
          type: io.kestra.plugin.core.execution.Fail
          errorMessage: "The difference {{ outputs.result.values.r | jq('.difference') |
            first }} does not equal the sum of unexplained items {{
            outputs.result.values.r | jq('.unexplained_sum') | first }}. A
            classification rule is wrong. Do not close."

    defaults:
      - id: block_close
        type: io.kestra.plugin.core.execution.Fail
        description: Unexplained items or a non-zero difference. The period stays open
          and the exceptions file says what to fix.
        errorMessage: >-
          Period {{ inputs.period }} not closed: difference {{
          outputs.result.values.r | jq('.difference') | first }}, {{
          outputs.result.values.r | jq('.unexplained') | first }} unexplained
          item(s): {{ outputs.result.values.r | jq('.unexplained_items') | first
          }}. {{ outputs.result.values.r | jq('.transposition_detail') | first
          }} See {{ render(vars.pack_dir) }}/exceptions.csv.