Microsoft Access Database Capacity & Ledger Integrity Sentinel

id: ms-access-database-capacity-and-integrity-sentinel
namespace: company.data

inputs:
  - id: database_path
    type: STRING
    defaults: "finance_ledger.accdb"
    description: Local or working directory filename for the target Microsoft Access
      database.

  - id: access_file_uri
    type: STRING
    defaults: ""
    description: Optional Kestra internal storage URI to an existing Access .accdb
      or .mdb database file.

  - id: max_unposted_threshold
    type: INT
    defaults: 10
    description: Maximum tolerable count of pending or unverified ledger
      transactions before raising an alert.

  - id: max_balance_discrepancy
    type: INT
    defaults: 0
    description: Maximum allowable numerical variance between total debits and total
      credits.

  - id: auto_flag_discrepancies
    type: BOOL
    defaults: false
    description: When true, updates unposted records to append an AUDIT_HOLD tag in
      the account code.

  - id: dry_run
    type: BOOL
    defaults: false
    description: When true, simulates remediation without modifying records in the
      Access database.

  - id: slack_channel
    type: STRING
    defaults: "#finance-ops-alerts"
    description: Slack destination channel for operational and reconciliation notifications.

triggers:
  - id: periodic_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 6 * * *"
    disabled: true
    inputs:
      database_path: "finance_ledger.accdb"
      access_file_uri: ""
      max_unposted_threshold: 10
      max_balance_discrepancy: 0
      auto_flag_discrepancies: false
      dry_run: false
      slack_channel: "#finance-ops-alerts"

tasks:
  - id: setup_test_database
    type: io.kestra.plugin.core.flow.If
    condition: "{{ inputs.access_file_uri == '' }}"
    then:
      - id: create_seed_ledger
        type: io.kestra.plugin.jdbc.access.Query
        url: "jdbc:ucanaccess:///{{ inputs.database_path }}"
        outputDbFile: true
        sql: |
          CREATE TABLE ledger_entries (
            entry_id AUTOINCREMENT PRIMARY KEY,
            account_code VARCHAR(30),
            debit DOUBLE,
            credit DOUBLE,
            posted BIT,
            entry_date TIMESTAMP
          );
          INSERT INTO ledger_entries (account_code, debit, credit, posted, entry_date) VALUES ('1010-CASH', 15000.00, 0.00, true, NOW());
          INSERT INTO ledger_entries (account_code, debit, credit, posted, entry_date) VALUES ('2010-AP', 0.00, 15000.00, true, NOW());
          INSERT INTO ledger_entries (account_code, debit, credit, posted, entry_date) VALUES ('1020-RECEIVABLES', 4500.00, 0.00, false, NOW());
          INSERT INTO ledger_entries (account_code, debit, credit, posted, entry_date) VALUES ('4010-REVENUE', 0.00, 4200.00, false, NOW());
        fetchType: NONE
        description: Bootstraps an operational test Access database with seeded
          transactions if no external file URI is provided.

  - id: probe_ledger_integrity
    type: io.kestra.plugin.jdbc.access.Query
    url: "jdbc:ucanaccess:///{{ inputs.database_path }}"
    accessFile: "{{ outputs.create_seed_ledger.databaseUri ??
      (inputs.access_file_uri != '' ? inputs.access_file_uri : null) }}"
    outputDbFile: true
    sql: |
      SELECT
        COUNT(*) AS total_entries,
        SUM(IIF(posted = false, 1, 0)) AS unposted_count,
        ROUND(SUM(debit), 2) AS total_debits,
        ROUND(SUM(credit), 2) AS total_credits,
        ROUND(ABS(SUM(debit) - SUM(credit)), 2) AS balance_delta
      FROM ledger_entries;
    fetchType: FETCH_ONE
    allowFailure: true
    description: Queries the Access database to audit ledger balances, unposted
      transaction counts, and debit/credit reconciliation parity.

  - id: classify_database_health
    type: io.kestra.plugin.core.debug.Return
    format: >-
      {{
        (outputs.probe_ledger_integrity is not defined or outputs.probe_ledger_integrity.row is not defined) ? 'PROBE_FAILED' :
        (outputs.probe_ledger_integrity.row.balance_delta > inputs.max_balance_discrepancy) ? 'CRITICAL_LEDGER_MISMATCH' :
        (outputs.probe_ledger_integrity.row.unposted_count >= inputs.max_unposted_threshold) ? 'WARNING_UNPOSTED_BACKLOG' :
        'HEALTHY_RECONCILED'
      }}
    description: Evaluates transaction audit metrics against financial governance
      and balance parity thresholds.

  - id: route_by_health_status
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ outputs.classify_database_health.value }}"
    cases:
      CRITICAL_LEDGER_MISMATCH:
        - id: alert_ledger_mismatch_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": "[CRITICAL] Microsoft Access Ledger Discrepancy Detected\n• Target Database: {{ inputs.database_path }}\n• Total Debits: ${{ outputs.probe_ledger_integrity.row.total_debits }}\n• Total Credits: ${{ outputs.probe_ledger_integrity.row.total_credits }}\n• Balance Delta: ${{ outputs.probe_ledger_integrity.row.balance_delta }} (Limit: ${{ inputs.max_balance_discrepancy }})\n• Unposted Entries: {{ outputs.probe_ledger_integrity.row.unposted_count }}\n• Execution ID: {{ execution.id }}\nAction Required: Debits and credits fail reconciliation parity. Audit unposted batches before monthly close."
            }

        - id: check_remediation
          type: io.kestra.plugin.core.flow.If
          condition: "{{ inputs.auto_flag_discrepancies == true and inputs.dry_run ==
            false }}"
          then:
            - id: flag_unbalanced_records
              type: io.kestra.plugin.jdbc.access.Query
              url: "jdbc:ucanaccess:///{{ inputs.database_path }}"
              accessFile: "{{ outputs.probe_ledger_integrity.databaseUri ??
                outputs.create_seed_ledger.databaseUri }}"
              outputDbFile: true
              sql: "UPDATE ledger_entries SET account_code = account_code & '-AUDIT_HOLD'
                WHERE posted = false;"
              fetchType: NONE

            - id: notify_remediation_slack
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              url: "{{ secret('SLACK_WEBHOOK_URL') }}"
              payload: |
                {
                  "channel": "{{ inputs.slack_channel }}",
                  "text": "[REMEDIATION] Unposted transactions have been marked with AUDIT_HOLD in {{ inputs.database_path }} to prevent premature posting."
                }

      WARNING_UNPOSTED_BACKLOG:
        - id: alert_backlog_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": "[WARNING] Microsoft Access Pending Transaction Backlog\n• Target Database: {{ inputs.database_path }}\n• Unposted Count: {{ outputs.probe_ledger_integrity.row.unposted_count }} (Threshold: {{ inputs.max_unposted_threshold }})\n• Execution ID: {{ execution.id }}\nBatch entries require operator posting approval."
            }

      HEALTHY_RECONCILED:
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          message: "Microsoft Access ledger healthy on {{ inputs.database_path }}. Debits
            (${{ outputs.probe_ledger_integrity.row.total_debits }}) and credits
            (${{ outputs.probe_ledger_integrity.row.total_credits }}) balance
            with zero delta."

      PROBE_FAILED:
        - id: alert_probe_failed_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": "[ALERT] Microsoft Access Database Connection Failure\n• Target File: {{ inputs.database_path }}\n• Execution ID: {{ execution.id }}\nUnable to connect to Access database via UCanAccess driver. Inspect file permissions and lock states."
            }
    description: Routes operational notifications and guarded remediation based on
      ledger reconciliation health.

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    message: "Access sentinel cycle {{ execution.id }} completed. Verdict: {{
      outputs.classify_database_health.value }}, Total rows: {{
      outputs.probe_ledger_integrity.row.total_entries ?? 0 }}, Balance delta:
      ${{ outputs.probe_ledger_integrity.row.balance_delta ?? 0 }}."
    description: Records an immutable audit log entry documenting execution metrics
      and remediation decisions.

errors:
  - id: alert_flow_error
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "channel": "{{ inputs.slack_channel }}",
        "text": "[ERROR] Microsoft Access Sentinel Workflow Failure\nFlow execution {{ execution.id }} encountered an unhandled error against {{ inputs.database_path }}. Review Kestra task logs."
      }