Rotation

async def rotate_encryption_keys(
    db_url: str,
    old_backend: EncryptionBackend,
    new_backend: EncryptionBackend,
) -> RotationResult:
    """Re-encrypt all session data from one backend to another.

    Reads all encrypted records from the four session tables
    (``sessions``, ``app_states``, ``user_states``, ``events``),
    identifies records encrypted with ``old_backend`` by checking the
    envelope ``backend_id`` byte, and re-encrypts them using
    ``new_backend``. Records already on a different backend are skipped
    silently.

    Uses ``update_time`` and the existing ciphertext value as an optimistic
    concurrency guard for ``sessions``, ``app_states``, and ``user_states``.
    If a record is modified between the rotation function's read and write
    (``rows_affected == 0``), it is counted as skipped. Run the function
    again to pick up remaining records.

    **Re-run semantics differ by rotation type:**

    - *Cross-backend rotation* (``old_backend.backend_id !=
      new_backend.backend_id``): Re-runs are safe. Already-rotated records
      carry ``new_backend.backend_id`` in their envelope and are skipped
      silently by the backend-id filter.
    - *Same-backend rotation* (``old_backend.backend_id ==
      new_backend.backend_id``, e.g., two ``FernetBackend`` instances):
      A single pass is expected. Re-running with the original ``old_backend``
      will attempt to decrypt already-rotated ciphertext with the old key
      and raise ``DecryptionError``. For same-backend rotation, stop or
      pause the service before running this function, run once, then
      reconfigure the service to use ``new_backend`` and restart.

    For the ``events`` table (no ``update_time`` column), ``rows_affected
    == 0`` means the event was cascade-deleted between read and write and
    is not counted as skipped.

    Cryptographic operations (``sync_decrypt``, ``sync_encrypt``) run per
    record in a thread via ``asyncio.to_thread()`` to avoid blocking the
    event loop. For databases with very large numbers of records, run this
    utility during a low-traffic window to minimise thread-pool pressure.

    Args:
        db_url: SQLAlchemy connection string (e.g.,
            ``"sqlite+aiosqlite:///sessions.db"``).
        old_backend: Backend used to decrypt existing records. Records
            whose envelope ``backend_id`` matches ``old_backend.backend_id``
            are re-encrypted.
        new_backend: Backend used to encrypt re-written records.

    Returns:
        A ``RotationResult`` with ``rotated`` (successfully re-encrypted)
        and ``skipped`` (concurrent-write collisions) counts across all
        four tables.

    Raises:
        DecryptionError: If a record cannot be decrypted with
            ``old_backend``, or if a stored value has a malformed
            envelope. Error messages never contain key material.

    Examples:
        Rotate from one Fernet passphrase to another:

        ```python
        from adk_secure_sessions import FernetBackend
        from adk_secure_sessions.rotation import rotate_encryption_keys

        old = FernetBackend("old-passphrase")
        new = FernetBackend("new-passphrase")
        result = await rotate_encryption_keys(
            db_url="sqlite+aiosqlite:///sessions.db",
            old_backend=old,
            new_backend=new,
        )
        print(f"Rotated {result.rotated} records, skipped {result.skipped}")
        ```

    See Also:
        [`adk_secure_sessions.services.encrypted_session`][]: Use
        ``additional_backends`` for cross-backend lazy migration
        (Path A, no utility required).
    """
    engine = create_async_engine(db_url)
    rotated = 0
    skipped = 0

    try:
        for spec in _TABLE_SPECS:
            async with engine.begin() as conn:
                r, s = await _rotate_table(
                    conn=conn,
                    table=spec["table"],
                    pk_cols=spec["pk_cols"],
                    enc_col=spec["enc_col"],
                    has_update_time=spec["has_update_time"],
                    old_backend=old_backend,
                    new_backend=new_backend,
                )
                rotated += r
                skipped += s
    finally:
        await engine.dispose()

    return RotationResult(rotated=rotated, skipped=skipped)