name: SQL rebuild audit hash-chain (per-tenant via postgres)

# Запускает per-tenant rebuild hash-chain для аудит-партиции через
# sudo -u postgres psql (обход limitation crm_supplier_worker роли —
# она не может SET session_replication_role).
#
# Поддерживает 2 таблицы (Stage 5 finding 1+2):
#   - activity_log → ROW(id,tenant_id,user_id,deal_id,event,old_value,
#     new_value,context,ip_address,user_agent,NULL::bytea,created_at)
#   - balance_transactions → ROW(id,tenant_id,type,amount_rub,amount_leads,
#     balance_rub_after,balance_leads_after,description,related_type,
#     related_id,user_id,admin_user_id,NULL::bytea,created_at)

on:
  workflow_dispatch:
    inputs:
      partition:
        description: 'Имя партиции, например activity_log_y2026_m05'
        required: true
        type: string
      from_id:
        description: 'ID с которого начать пересчёт (включительно)'
        required: true
        type: string
      table_kind:
        description: 'activity_log | balance_transactions | pd_processing_log | tenant_operations_log'
        required: true
        type: choice
        options:
          - activity_log
          - balance_transactions
          - pd_processing_log
          - tenant_operations_log
      confirm_apply:
        description: 'Подтверждаю выполнение mutating cleanup'
        required: true
        default: false
        type: boolean

jobs:
  rebuild:
    runs-on: ubuntu-latest
    timeout-minutes: 10

    env:
      LIDERRA_HOST: 111.88.246.137
      LIDERRA_USER: ubuntu
      PARTITION: ${{ github.event.inputs.partition }}
      FROM_ID: ${{ github.event.inputs.from_id }}
      TABLE_KIND: ${{ github.event.inputs.table_kind }}

    steps:
      - name: Confirm check
        run: |
          if [[ "${{ github.event.inputs.confirm_apply }}" != "true" ]]; then
            echo "::error::confirm_apply=true обязателен"
            exit 1
          fi
          # Sanity: partition must match table_kind
          case "$TABLE_KIND" in
            activity_log)
              if [[ ! "$PARTITION" =~ ^activity_log_y[0-9]{4}_m[0-9]{2}$ ]]; then
                echo "::error::partition '$PARTITION' не соответствует table_kind=activity_log"
                exit 1
              fi
              ;;
            balance_transactions)
              if [[ ! "$PARTITION" =~ ^balance_transactions_y[0-9]{4}_m[0-9]{2}$ ]]; then
                echo "::error::partition '$PARTITION' не соответствует table_kind=balance_transactions"
                exit 1
              fi
              ;;
            pd_processing_log)
              if [[ ! "$PARTITION" =~ ^pd_processing_log_y[0-9]{4}_m[0-9]{2}$ ]]; then
                echo "::error::partition '$PARTITION' не соответствует table_kind=pd_processing_log"
                exit 1
              fi
              ;;
            tenant_operations_log)
              if [[ ! "$PARTITION" =~ ^tenant_operations_log_y[0-9]{4}_m[0-9]{2}$ ]]; then
                echo "::error::partition '$PARTITION' не соответствует table_kind=tenant_operations_log"
                exit 1
              fi
              ;;
            *)
              echo "::error::table_kind unknown"
              exit 1
              ;;
          esac
          if ! [[ "$FROM_ID" =~ ^[0-9]+$ ]]; then
            echo "::error::from_id must be numeric"
            exit 1
          fi

      - name: Setup SSH key
        run: |
          mkdir -p ~/.ssh
          echo "${{ secrets.LIDERRA_SSH_KEY }}" > ~/.ssh/liderra_deploy
          chmod 600 ~/.ssh/liderra_deploy
          ssh-keyscan -H ${{ env.LIDERRA_HOST }} >> ~/.ssh/known_hosts 2>/dev/null

      - name: Execute SQL rebuild on prod
        run: |
          # Build ROW expression per table_kind (mirror AuditChainConfig::TABLES)
          case "$TABLE_KIND" in
            activity_log)
              ROW_EXPR="ROW(t.id, t.tenant_id, t.user_id, t.deal_id, t.event, t.old_value, t.new_value, t.context, t.ip_address, t.user_agent, NULL::bytea, t.created_at)"
              ;;
            balance_transactions)
              ROW_EXPR="ROW(t.id, t.tenant_id, t.type, t.amount_rub, t.amount_leads, t.balance_rub_after, t.balance_leads_after, t.description, t.related_type, t.related_id, t.user_id, t.admin_user_id, NULL::bytea, t.created_at)"
              ;;
            pd_processing_log)
              ROW_EXPR="ROW(t.id, t.tenant_id, t.subject_type, t.subject_id, t.action, t.purpose, t.actor_tenant_user_id, t.actor_admin_user_id, t.ip_address, NULL::bytea, t.created_at)"
              ;;
            tenant_operations_log)
              ROW_EXPR="ROW(t.id, t.tenant_id, t.user_id, t.entity_type, t.entity_id, t.event, t.payload_before, t.payload_after, t.ip_address, t.user_agent, NULL::bytea, t.created_at)"
              ;;
          esac

          # Build SQL with substituted PARTITION + FROM_ID + ROW_EXPR
          cat > /tmp/rebuild.sql <<SQL
          \\set ON_ERROR_STOP 1

          SELECT 'BEFORE: mismatches in partition' AS phase, COUNT(*) AS cnt
          FROM (
            WITH ordered AS (
              SELECT id, tenant_id, log_hash AS stored_hash,
                     LAG(log_hash) OVER (PARTITION BY tenant_id ORDER BY id) AS prev_hash
              FROM ${PARTITION}
            )
            SELECT o.id
            FROM ordered o
            WHERE o.stored_hash IS DISTINCT FROM
              digest(
                COALESCE(o.prev_hash, ''::bytea)
                || (SELECT ${ROW_EXPR}::text::bytea FROM ${PARTITION} t WHERE t.id = o.id),
                'sha256'
              )
          ) sub;

          DO \$\$
          DECLARE
            tenant_rec RECORD;
            row_rec RECORD;
            prev_hash BYTEA;
            new_hash BYTEA;
            updated_count INT := 0;
            tenant_count INT := 0;
          BEGIN
            SET session_replication_role = 'replica';

            FOR tenant_rec IN
              SELECT DISTINCT tenant_id FROM ${PARTITION} WHERE id >= ${FROM_ID} ORDER BY tenant_id
            LOOP
              tenant_count := tenant_count + 1;

              SELECT log_hash INTO prev_hash
              FROM ${PARTITION}
              WHERE tenant_id = tenant_rec.tenant_id AND id < ${FROM_ID}
              ORDER BY id DESC LIMIT 1;

              FOR row_rec IN
                SELECT id FROM ${PARTITION}
                WHERE tenant_id = tenant_rec.tenant_id AND id >= ${FROM_ID}
                ORDER BY id
              LOOP
                UPDATE ${PARTITION} p
                SET log_hash = digest(
                  COALESCE(prev_hash, ''::bytea)
                  || (SELECT ${ROW_EXPR}::text::bytea FROM ${PARTITION} t WHERE t.id = row_rec.id),
                  'sha256'
                )
                WHERE p.id = row_rec.id
                RETURNING log_hash INTO new_hash;

                prev_hash := new_hash;
                updated_count := updated_count + 1;
              END LOOP;
            END LOOP;

            SET session_replication_role = 'origin';
            RAISE NOTICE 'Rebuild complete: % tenants, % rows updated', tenant_count, updated_count;
          END\$\$;

          SELECT 'AFTER: mismatches in partition' AS phase, COUNT(*) AS cnt
          FROM (
            WITH ordered AS (
              SELECT id, tenant_id, log_hash AS stored_hash,
                     LAG(log_hash) OVER (PARTITION BY tenant_id ORDER BY id) AS prev_hash
              FROM ${PARTITION}
            )
            SELECT o.id
            FROM ordered o
            WHERE o.stored_hash IS DISTINCT FROM
              digest(
                COALESCE(o.prev_hash, ''::bytea)
                || (SELECT ${ROW_EXPR}::text::bytea FROM ${PARTITION} t WHERE t.id = o.id),
                'sha256'
              )
          ) sub;
          SQL

          scp -i ~/.ssh/liderra_deploy /tmp/rebuild.sql ${{ env.LIDERRA_USER }}@${{ env.LIDERRA_HOST }}:/tmp/rebuild.sql
          ssh -i ~/.ssh/liderra_deploy ${{ env.LIDERRA_USER }}@${{ env.LIDERRA_HOST }} 'sudo -u postgres psql -d liderra -f /tmp/rebuild.sql && rm /tmp/rebuild.sql'

      - name: Cleanup SSH key
        if: always()
        run: rm -f ~/.ssh/liderra_deploy
