gbfs-compact.yml

name: GBFS Daily Compaction
on:
  schedule:
    - cron: '15 0 * * *'  # 00:15 UTC, after all per-minute WALs for yesterday are written
  workflow_dispatch:
    inputs:
      date:
        description: 'Date to compact (YYYY-MM-DD, default: yesterday)'
        required: false

jobs:
  compact:
    runs-on: ubuntu-latest
    env:
      AWS_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
      AWS_SECRET_ACCESS_KEY: ${{ secrets.R2_SECRET_ACCESS_KEY }}
      AWS_ENDPOINT_URL: https://${{ secrets.CLOUDFLARE_ACCOUNT_ID }}.r2.cloudflarestorage.com
      AWS_DEFAULT_REGION: auto
    steps:
      - uses: actions/checkout@v5

      - uses: astral-sh/setup-uv@v7
        with:
          python-version: '3.12'
          enable-cache: true

      - name: Sync deps + activate venv
        run: |
          uv sync --frozen
          echo "$GITHUB_WORKSPACE/.venv/bin" >> "$GITHUB_PATH"

      - name: Determine date
        id: date
        run: |
          if [ -n "${{ inputs.date }}" ]; then
            DATE="${{ inputs.date }}"
          else
            DATE=$(date -u -d yesterday +%Y-%m-%d)
          fi
          echo "date=$DATE" >> "$GITHUB_OUTPUT"
          # Trigger /d1 build on the 1st of each month (yesterday = last day of prev month)
          DOM=$(date -u -d "$DATE +1 day" +%d)
          echo "trigger_d1=$([ "$DOM" = "01" ] && echo true || echo false)" >> "$GITHUB_OUTPUT"
          # Trigger /mo1 build on Jan 1 (yesterday = Dec 31)
          NEXT_MONTH=$(date -u -d "$DATE +1 day" +%m)
          echo "trigger_mo1=$([ "$NEXT_MONTH" = "01" ] && [ "$DOM" = "01" ] && echo true || echo false)" >> "$GITHUB_OUTPUT"
          echo "ym=$(echo $DATE | cut -c 1-7)" >> "$GITHUB_OUTPUT"
          echo "year=$(echo $DATE | cut -c 1-4)" >> "$GITHUB_OUTPUT"

      - name: Compact WAL → parquet
        run: python3 gbfs/compact-r2.py all ${{ steps.date.outputs.date }}

      # Empty/full station bitmaps (`specs/avail-empty-bitmaps.md`): the day's four
      # planes → `empty-v1/` Zarr day-shards, from the daily parquet just written.
      # `-V` range-reads one hour of every shard back through the reference reader.
      - name: Build empty-v1 bitmaps for the day
        env:
          R2_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
          R2_SECRET_ACCESS_KEY: ${{ secrets.R2_SECRET_ACCESS_KEY }}
          CLOUDFLARE_ACCOUNT_ID: ${{ secrets.CLOUDFLARE_ACCOUNT_ID }}
        run: ctbk gbfs empty build -C -V ${{ steps.date.outputs.date }}

      # SMG source (`specs/avail-smg-pyramid.md`): `empty build` above also wrote the
      # day's `gbfs/smg/<day>.parquet`. This self-heals any day that has a status
      # parquet but no SMG parquet (a missed run, or days compacted before the SMG
      # step existed) — a no-op when nothing is missing — so the fill below never
      # sees a real source hole.
      - name: Build any missing gbfs/smg day parquets
        env:
          R2_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
          R2_SECRET_ACCESS_KEY: ${{ secrets.R2_SECRET_ACCESS_KEY }}
          CLOUDFLARE_ACCOUNT_ID: ${{ secrets.CLOUDFLARE_ACCOUNT_ID }}
        run: ctbk gbfs smg backfill -C -k -t ${{ steps.date.outputs.date }}

      # SMG pyramid: a declarative Batch gap-fill (`-f`) builds only the rungs the new
      # day completes, into the REAL prefix (`-p smg-v1` — the default would be the
      # `-engine-check` scratch prefix). Own AWS env: the job-level `AWS_*` above
      # point at R2, and Batch is real AWS (us-east-1). `-W` tails the job so a
      # failed fill fails this step.
      # Uncapped range (genesis → now): the base image (pyrmts ≥ 22a5f32) DEFERS the
      # trailing rungs over today's not-yet-existing source parquet (an open period)
      # instead of writing them as 0-row shards — nothing to sweep, exit 0, and the
      # next day's run builds them. An absent CLOSED day (a compaction that never
      # ran) fails this step BEFORE anything is written (pyrmts ≥ a174254, exit 4);
      # the `smg backfill` step above is what heals that, so a failure here means
      # the day has no status parquet either. Memory budget: the submitter (pyrmts ≥ c594d6b)
      # pins `-b` to 70% of the job definition's memory, so the engine's in-container
      # detection (which misreads Fargate's host memory) is never consulted.
      - name: Fill smg-v1 pyramid (Batch)
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          AWS_ENDPOINT_URL: ''
          AWS_DEFAULT_REGION: us-east-1
          R2_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
          R2_SECRET_ACCESS_KEY: ${{ secrets.R2_SECRET_ACCESS_KEY }}
          CLOUDFLARE_ACCOUNT_ID: ${{ secrets.CLOUDFLARE_ACCOUNT_ID }}
          CTBK_REGISTRY_SECRET: ${{ secrets.CTBK_REGISTRY_SECRET }}
        run: ctbk gbfs engine submit -C smg-v1 -p smg-v1 -x ctbk_engine_src:smg_daily -f -W -w 1h

      - name: Build /h1 avail-agg for the day
        run: |
          python3 -c "from ctbk.avail_agg import AvailAggH1Day; AvailAggH1Day('${{ steps.date.outputs.date }}').create()"
          aws s3 cp "r2/ctbk/avail/agg/h1/${{ steps.date.outputs.date }}.parquet" \
                    "s3://ctbk/avail/agg/h1/${{ steps.date.outputs.date }}.parquet" \
                    --endpoint-url "$AWS_ENDPOINT_URL"

      - name: Build /day raw bundle for the day
        run: |
          python3 -c "from ctbk.avail_raw_day import AvailRawDay; AvailRawDay('${{ steps.date.outputs.date }}').create()"
          aws s3 cp "r2/ctbk/gbfs/avail/raw/day/${{ steps.date.outputs.date }}.parquet" \
                    "s3://ctbk/gbfs/avail/raw/day/${{ steps.date.outputs.date }}.parquet" \
                    --endpoint-url "$AWS_ENDPOINT_URL"

      - name: Build /d1 avail-agg for closed month
        if: ${{ steps.date.outputs.trigger_d1 == 'true' }}
        run: |
          python3 -c "from ctbk.avail_agg import AvailAggD1Month; AvailAggD1Month('${{ steps.date.outputs.ym }}').create()"
          aws s3 cp "r2/ctbk/avail/agg/d1/${{ steps.date.outputs.ym }}.parquet" \
                    "s3://ctbk/avail/agg/d1/${{ steps.date.outputs.ym }}.parquet" \
                    --endpoint-url "$AWS_ENDPOINT_URL"

      - name: Build /mo1 avail-agg for closed year
        if: ${{ steps.date.outputs.trigger_mo1 == 'true' }}
        run: |
          python3 -c "from ctbk.avail_agg import AvailAggMo1Year; AvailAggMo1Year(${{ steps.date.outputs.year }}).create()"
          aws s3 cp "r2/ctbk/avail/agg/mo1/${{ steps.date.outputs.year }}.parquet" \
                    "s3://ctbk/avail/agg/mo1/${{ steps.date.outputs.year }}.parquet" \
                    --endpoint-url "$AWS_ENDPOINT_URL"