spirekeeper/migrations.py

# Satoshi Machine v2 — single squashed migration.
#
# History note: m001-m004 were the legacy Lamassu schema; m005-m007 staged
# the v2 redesign (initial schema → payment_hash idempotency fix → notes
# column → concurrency claim + wallet UNIQUE index). Collapsed back into a
# single m001 during the v2-bitspire development branch since no production
# data was affected and the staged sequence had a SQLite CREATE-INDEX
# syntax bug. The pre-collapse history is preserved in git on commits
# prior to the collapse.
#
# Installs upgrading from the v1 Lamassu schema must uninstall + reinstall
# the extension to reset the LNbits dbversions tracker. The DROP TABLE
# IF EXISTS at the top of m001 also cleans the v1 tables if they happen
# to survive a partial wipe.


async def m001_satmachine_v2_initial(db):
    """Single-shot v2 schema for the Satoshi Machine admin extension.

    Drops every legacy Lamassu table (lamassu_config, lamassu_transactions,
    plus the singular-config v1 dca_clients/deposits/payments) and creates
    the v2 multi-tenant schema:

      - super_config:           singleton platform-fee config (super only)
      - dca_machines:           per-operator multi-machine registry by npub
      - dca_clients:            LP registrations scoped per (machine, user)
      - dca_deposits:           fiat the operator records against an LP
      - dca_settlements:        bitSpire kind-21000 idempotency table
      - dca_commission_splits:  operator's remainder-distribution rules
      - dca_payments:           leg-typed distribution audit trail
      - dca_telemetry:          sparse kind-30078/30079 snapshots per machine

    CRITICAL design choices (preserved from the staged migrations):
      * payment_hash is the UNIQUE idempotency key on dca_settlements
        (LN payment_hash is globally unique and always present at the
        Payment layer — fix for the original "use bitspire_event_id"
        false start).
      * platform_fee_sats + operator_fee_sats stored as absolute BIGINT
        (not derived percentages). The contract is locked at landing time;
        post-v1 customer-discount engine writes here without a migration.
      * dca_machines.wallet_id UNIQUE — defence-in-depth against the
        wallet-IDOR funds-theft vector (the API layer also checks
        wallet ownership; the index is the second line of defence).
      * processing_claim on dca_settlements — optimistic-lock token for
        concurrent process_settlement invocations.
      * notes on dca_settlements — append-only audit memo for partial-
        dispense recompute + operator-authored notes (see
        aiolabs/spirekeeper#6 for the future structured audit table).
    """
    # 1. Drop legacy v1 tables. IF EXISTS handles both fresh-install
    #    paths (no-op) and migration from a v1 schema (cleans up).
    for table in (
        "lamassu_transactions",
        "lamassu_config",
        "dca_payments",
        "dca_deposits",
        "dca_clients",
    ):
        await db.execute(f"DROP TABLE IF EXISTS spirekeeper.{table}")

    # 2. super_config — singleton (id='default') with platform-fee config.
    await db.execute(f"""
        CREATE TABLE IF NOT EXISTS spirekeeper.super_config (
            id TEXT PRIMARY KEY,
            super_fee_fraction DECIMAL(10,4) NOT NULL DEFAULT 0.0000,
            super_fee_wallet_id TEXT,
            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
        );
        """)
    existing = await db.fetchone(
        "SELECT id FROM spirekeeper.super_config WHERE id = 'default'"
    )
    if not existing:
        await db.execute(
            "INSERT INTO spirekeeper.super_config (id, super_fee_fraction) "
            "VALUES ('default', 0.0000)"
        )

    # 3. dca_machines — one row per bitSpire ATM, owned by exactly one
    #    operator. wallet_id UNIQUE prevents the IDOR funds-theft vector.
    await db.execute(f"""
        CREATE TABLE IF NOT EXISTS spirekeeper.dca_machines (
            id TEXT PRIMARY KEY,
            operator_user_id TEXT NOT NULL,
            machine_npub TEXT NOT NULL UNIQUE,
            wallet_id TEXT NOT NULL,
            name TEXT,
            location TEXT,
            fiat_code TEXT NOT NULL DEFAULT 'GTQ',
            is_active BOOLEAN NOT NULL DEFAULT true,
            created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
        );
        """)
    await db.execute(
        "CREATE INDEX IF NOT EXISTS dca_machines_operator_idx "
        "ON dca_machines (operator_user_id)"
    )
    await db.execute(
        "CREATE UNIQUE INDEX IF NOT EXISTS dca_machines_wallet_id_uq "
        "ON dca_machines (wallet_id)"
    )

    # 4. dca_clients — per-(machine, LP) registrations. Pure machine
    #    enrolment record: no wallet, no mode, no autoforward — those are
    #    LP-controlled at the user level via dca_lp (see below). Operator
    #    just decides "this LP is enrolled at my machine"; everything
    #    delivery-related is the LP's own preference.
    await db.execute(f"""
        CREATE TABLE IF NOT EXISTS spirekeeper.dca_clients (
            id TEXT PRIMARY KEY,
            machine_id TEXT NOT NULL,
            user_id TEXT NOT NULL,
            username TEXT,
            status TEXT NOT NULL DEFAULT 'active',
            created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
        );
        """)
    await db.execute(
        "CREATE UNIQUE INDEX IF NOT EXISTS dca_clients_machine_user_uq "
        "ON dca_clients (machine_id, user_id)"
    )
    await db.execute(
        "CREATE INDEX IF NOT EXISTS dca_clients_user_idx ON dca_clients (user_id)"
    )

    # 4a. dca_lp — LP-level (per-user) DCA preferences. ONE row per LNbits
    #     user that has onboarded as a Liquidity Provider, regardless of
    #     how many machines they're enrolled at. Owned by the LP (writes
    #     come from the satmachineclient extension under the LP's session),
    #     read by spirekeeper during distribution to resolve "where do
    #     DCA payouts for this LP go?"
    #
    #     Gating: spirekeeper refuses to create deposits for an LP who
    #     doesn't have a dca_lp row yet. The LP must onboard via
    #     satmachineclient first (which auto-creates the row with their
    #     default LNbits wallet on first dashboard visit). Forces every
    #     LP through a "yes, I am here and this is where I want my sats"
    #     gesture before any fiat starts accumulating against them.
    await db.execute(f"""
        CREATE TABLE IF NOT EXISTS spirekeeper.dca_lp (
            user_id TEXT PRIMARY KEY,
            dca_wallet_id TEXT NOT NULL,
            default_dca_mode TEXT NOT NULL DEFAULT 'flow',
            fixed_mode_daily_limit DECIMAL(10,2),
            autoforward_ln_address TEXT,
            autoforward_enabled BOOLEAN NOT NULL DEFAULT false,
            created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
        );
        """)

    # 5. dca_deposits — fiat the operator (or super) records against an LP
    #    at a machine. creator_user_id preserves audit trail.
    await db.execute(f"""
        CREATE TABLE IF NOT EXISTS spirekeeper.dca_deposits (
            id TEXT PRIMARY KEY,
            client_id TEXT NOT NULL,
            machine_id TEXT NOT NULL,
            creator_user_id TEXT NOT NULL,
            amount DECIMAL(10,2) NOT NULL,
            currency TEXT NOT NULL DEFAULT 'GTQ',
            status TEXT NOT NULL DEFAULT 'pending',
            notes TEXT,
            created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
            confirmed_at TIMESTAMP
        );
        """)
    await db.execute(
        "CREATE INDEX IF NOT EXISTS dca_deposits_client_idx "
        "ON dca_deposits (client_id, created_at DESC)"
    )

    # 6. dca_settlements — idempotency table for bitSpire-driven settlements.
    #    payment_hash UNIQUE handles subscription replays + dispatcher
    #    double-fires. processing_claim is the optimistic-lock token
    #    written by claim_settlement_for_processing. notes is the
    #    append-only audit memo for partial-dispense + operator notes.
    #
    #    platform_fee_sats and operator_fee_sats are absolute BIGINT,
    #    NOT derived fractions — when the v2 customer-discount engine
    #    ships, these two columns are the audit-grade record of who
    #    forgave what per transaction. Do not collapse them into a single
    #    fee_fraction. See plan section "Customer discounts" and #10.
    await db.execute(f"""
        CREATE TABLE IF NOT EXISTS spirekeeper.dca_settlements (
            id TEXT PRIMARY KEY,
            machine_id TEXT NOT NULL,
            payment_hash TEXT NOT NULL UNIQUE,
            bitspire_event_id TEXT,
            bitspire_txid TEXT,
            wire_sats BIGINT NOT NULL,
            fiat_amount DECIMAL(10,2) NOT NULL,
            fiat_code TEXT NOT NULL DEFAULT 'GTQ',
            exchange_rate REAL NOT NULL,
            principal_sats BIGINT NOT NULL,
            fee_sats BIGINT NOT NULL,
            platform_fee_sats BIGINT NOT NULL,
            operator_fee_sats BIGINT NOT NULL,
            tx_type TEXT NOT NULL,
            bills_json TEXT,
            cassettes_json TEXT,
            status TEXT NOT NULL DEFAULT 'pending',
            error_message TEXT,
            processed_at TIMESTAMP,
            created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
            notes TEXT,
            processing_claim TEXT
        );
        """)
    await db.execute(
        "CREATE INDEX IF NOT EXISTS dca_settlements_machine_idx "
        "ON dca_settlements (machine_id, created_at DESC)"
    )

    # 7. dca_commission_splits — operator's rules for distributing the
    #    *remainder* (fee_sats - platform_fee_sats). One row per
    #    leg. machine_id=NULL = operator default; non-null = per-machine
    #    override. Sum(fraction) per (operator, machine) must equal 1.0 —
    #    enforced at write-time in crud.py.
    #
    #    `target` accepts any of (splitpayments-style):
    #      - LNbits wallet id (UUID-shaped)
    #      - LNbits wallet invoice key (resolved via get_wallet_for_key)
    #      - Lightning address (user@domain)
    #      - LNURL string (bech32 LNURL...)
    #    Resolution lives in distribution._pay_one_split_leg.
    await db.execute(f"""
        CREATE TABLE IF NOT EXISTS spirekeeper.dca_commission_splits (
            id TEXT PRIMARY KEY,
            machine_id TEXT,
            operator_user_id TEXT NOT NULL,
            target TEXT NOT NULL,
            label TEXT,
            fraction DECIMAL(10,4) NOT NULL,
            sort_order INTEGER NOT NULL DEFAULT 0,
            created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
        );
        """)
    await db.execute(
        "CREATE INDEX IF NOT EXISTS dca_commission_splits_lookup_idx "
        "ON dca_commission_splits (operator_user_id, machine_id)"
    )

    # 8. dca_payments — every leg of every distribution. leg_type
    #    discriminator: dca | super_fee | operator_split | settlement |
    #    autoforward | refund. status enum: pending | completed | failed |
    #    voided | skipped | refunded.
    await db.execute(f"""
        CREATE TABLE IF NOT EXISTS spirekeeper.dca_payments (
            id TEXT PRIMARY KEY,
            settlement_id TEXT,
            client_id TEXT,
            machine_id TEXT NOT NULL,
            operator_user_id TEXT NOT NULL,
            leg_type TEXT NOT NULL,
            destination_wallet_id TEXT,
            destination_ln_address TEXT,
            amount_sats BIGINT NOT NULL,
            amount_fiat DECIMAL(10,2),
            exchange_rate REAL,
            transaction_time TIMESTAMP NOT NULL,
            external_payment_hash TEXT,
            status TEXT NOT NULL DEFAULT 'pending',
            error_message TEXT,
            created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
        );
        """)
    await db.execute(
        "CREATE INDEX IF NOT EXISTS dca_payments_client_idx "
        "ON dca_payments (client_id, created_at DESC)"
    )
    await db.execute(
        "CREATE INDEX IF NOT EXISTS dca_payments_settlement_idx "
        "ON dca_payments (settlement_id)"
    )
    await db.execute(
        "CREATE INDEX IF NOT EXISTS dca_payments_operator_idx "
        "ON dca_payments (operator_user_id, leg_type)"
    )

    # 9. dca_telemetry — latest replaceable kind-30078 (public availability
    #    beacon) and kind-30079 (operator-only fleet telemetry) snapshots
    #    per machine. The beacon today (lamassu-next/dev @ 2b712af) ships
    #    only cash_in/cash_out/cash_level/fiat/model — post-#43 fields
    #    (name, location, geo, fees, limits, denominations, version) are
    #    nullable until that upstream issue lands. Ingest opportunistically.
    await db.execute("""
        CREATE TABLE IF NOT EXISTS spirekeeper.dca_telemetry (
            machine_id TEXT PRIMARY KEY,
            beacon_cash_in BOOLEAN,
            beacon_cash_out BOOLEAN,
            beacon_cash_level TEXT,
            beacon_fiat TEXT,
            beacon_model TEXT,
            beacon_name TEXT,
            beacon_location TEXT,
            beacon_geo TEXT,
            beacon_fees_json TEXT,
            beacon_limits_json TEXT,
            beacon_denominations_json TEXT,
            beacon_version TEXT,
            beacon_received_at TIMESTAMP,
            telemetry_json TEXT,
            telemetry_received_at TIMESTAMP
        );
        """)


async def m002_rename_commission_split_wallet_id_to_target(db):
    """One-off correction for installs whose `dca_commission_splits` table
    pre-exists from an earlier partial v2 migration run (where the column
    was named `wallet_id`). The collapsed m001 uses `CREATE TABLE IF NOT
    EXISTS`, which is a no-op when the table already exists — so the
    schema drift survives the documented uninstall + reinstall workflow
    because LNbits' uninstall wipes the dbversions tracker but NOT the
    spirekeeper.sqlite3 file on disk.

    Idempotent: probes for the `wallet_id` column via a SELECT. If the
    probe succeeds the column still exists and we RENAME it; otherwise
    the rename is already done (or the table was fresh) and we no-op.

    Fresh installs from m001 onward already have `target` directly — for
    them this migration is a no-op.
    """
    try:
        await db.fetchone(
            "SELECT wallet_id FROM spirekeeper.dca_commission_splits LIMIT 1"
        )
    except Exception:
        # wallet_id column doesn't exist; either m001 produced the correct
        # schema on a fresh install or the rename already landed.
        return
    await db.execute(
        "ALTER TABLE spirekeeper.dca_commission_splits "
        "RENAME COLUMN wallet_id TO target"
    )


async def m003_rename_settlements_net_sats_to_principal_sats(db):
    """Rename `dca_settlements.net_sats` → `principal_sats` for clarity.

    "Net" in financial accounting is overloaded (net of what?). In the
    bitSpire/DCA context this column is specifically the principal the
    operator distributes to LPs (gross − commission), not a generic
    "net" amount. Renaming locally before any bitSpire firmware locks
    the wire-level name; lamassu-next#44 should adopt the same name.

    Idempotent: probes for the old `net_sats` column. If present, rename.
    """
    try:
        await db.fetchone("SELECT net_sats FROM spirekeeper.dca_settlements LIMIT 1")
    except Exception:
        return
    await db.execute(
        "ALTER TABLE spirekeeper.dca_settlements "
        "RENAME COLUMN net_sats TO principal_sats"
    )


async def m004_introduce_dca_lp_table(db):
    """Hoist LP-level state (wallet, mode, autoforward) out of dca_clients
    into a per-user dca_lp table. dca_clients becomes a pure (machine, LP)
    enrolment record; everything delivery-related becomes the LP's own
    preference, owned and written by satmachineclient.

    Why: the per-row state on dca_clients was a denormalised duplicate of
    user-level intent ("which wallet should my DCA land in?" + "should it
    forward to my LN address?" — same answer regardless of which machine
    paid). Today's update_lp_autoforward already does a multi-row UPDATE
    to keep the rows in sync — a smell of state belonging one level up.

    Fresh installs from m001 onward land on the new schema directly.
    Existing installs (pre-m004 test data) get migrated here:
      1. Create dca_lp table (no-op if already present from m001 path).
      2. Backfill dca_lp from existing dca_clients rows, picking the
         most-recently-updated row per user_id when an LP is enrolled at
         multiple machines.
      3. Drop the moved columns from dca_clients.

    Idempotent: probes for the legacy `dca_clients.wallet_id` column. If
    absent the install already on the new shape; no-op.
    """
    try:
        await db.fetchone("SELECT wallet_id FROM spirekeeper.dca_clients LIMIT 1")
    except Exception:
        return

    # Step 1: create dca_lp if it doesn't exist yet. m001 on a fresh install
    # already created it; on a pre-m004 install we're creating it here.
    await db.execute(f"""
        CREATE TABLE IF NOT EXISTS spirekeeper.dca_lp (
            user_id TEXT PRIMARY KEY,
            dca_wallet_id TEXT NOT NULL,
            default_dca_mode TEXT NOT NULL DEFAULT 'flow',
            fixed_mode_daily_limit DECIMAL(10,2),
            autoforward_ln_address TEXT,
            autoforward_enabled BOOLEAN NOT NULL DEFAULT false,
            created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
            updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
        );
        """)

    # Step 2: backfill dca_lp from dca_clients. Pick the latest row per
    # user (by updated_at, falling back to created_at) when the LP is
    # enrolled at multiple machines — that row reflects their most
    # recent intent. ROW_NUMBER() OVER (...) requires SQLite 3.25+ (2018).
    await db.execute("""
        INSERT OR IGNORE INTO spirekeeper.dca_lp
            (user_id, dca_wallet_id, default_dca_mode, fixed_mode_daily_limit,
             autoforward_ln_address, autoforward_enabled,
             created_at, updated_at)
        SELECT user_id, wallet_id, dca_mode, fixed_mode_daily_limit,
               autoforward_ln_address, autoforward_enabled,
               created_at, updated_at
        FROM (
            SELECT *, ROW_NUMBER() OVER (
                PARTITION BY user_id
                ORDER BY updated_at DESC, created_at DESC
            ) AS rn
            FROM spirekeeper.dca_clients
        ) ranked
        WHERE rn = 1
        """)

    # Step 3: drop the moved columns from dca_clients. ALTER TABLE DROP
    # COLUMN needs SQLite 3.35+ (2021). One column per ALTER (SQLite
    # doesn't support multi-column DROP).
    for col in (
        "wallet_id",
        "dca_mode",
        "fixed_mode_daily_limit",
        "autoforward_ln_address",
        "autoforward_enabled",
    ):
        await db.execute(f"ALTER TABLE spirekeeper.dca_clients DROP COLUMN {col}")


async def m006_rename_to_canonical_sat_vocabulary(db):
    """Adopt the cross-codebase canonical sat-amount vocabulary AND drop
    the now-obsolete Lamassu-era fallback columns, per the decision at
    memory `reference_sat_amount_vocabulary.md` (2026-05-26):

    Renames:
      - dca_settlements.gross_sats      → wire_sats
      - dca_settlements.commission_sats → fee_sats
      - super_config.super_fee_pct      → super_fee_fraction
      - dca_commission_splits.pct       → fraction

    Drops (Lamassu-era reverse-derivation is obsolete since bitSpire
    stamps both `principal_sats` AND `fee_sats` directly on
    Payment.extra per lamassu-next#44 — there's nothing to back-derive):
      - dca_machines.fallback_commission_pct (was the rate used by the
        deleted `_parse_fallback` path)
      - dca_settlements.used_fallback_split (was the per-row marker for
        that path)

    Same canonical applies on the lamassu-next + atm-tui side; the
    rename is coordinated via `~/dev/coordination/log.md` (2026-05-26).

    Each step is idempotent — probe for the OLD column; rename/drop only
    if present; otherwise no-op (covers fresh installs where m001
    already laid down the canonical schema).

    Why a single migration: all driven by the same decision and any
    external code wants to see the whole rename + cleanup land at once.
    """
    renames = [
        ("dca_settlements", "gross_sats", "wire_sats"),
        ("dca_settlements", "commission_sats", "fee_sats"),
        ("super_config", "super_fee_pct", "super_fee_fraction"),
        ("dca_commission_splits", "pct", "fraction"),
    ]
    for table, old_col, new_col in renames:
        try:
            await db.fetchone(f"SELECT {old_col} FROM spirekeeper.{table} LIMIT 1")
        except Exception:
            # old column doesn't exist; either rename already landed or
            # m001 produced the canonical schema directly on fresh install.
            continue
        await db.execute(
            f"ALTER TABLE spirekeeper.{table} "
            f"RENAME COLUMN {old_col} TO {new_col}"
        )

    # Drop the Lamassu-era fallback columns. Same idempotency pattern.
    # Try both old (_pct) and new (_fraction) names for the dca_machines
    # column since an install could be at either rename state.
    drops = [
        ("dca_machines", "fallback_commission_pct"),
        ("dca_machines", "fallback_commission_fraction"),
        ("dca_settlements", "used_fallback_split"),
    ]
    for table, col in drops:
        try:
            await db.fetchone(f"SELECT {col} FROM spirekeeper.{table} LIMIT 1")
        except Exception:
            # column doesn't exist; either already dropped or never present.
            continue
        await db.execute(f"ALTER TABLE spirekeeper.{table} DROP COLUMN {col}")


async def m005_lock_deposit_currency_to_machine_fiat_code(db):
    """Rewrite every `dca_deposits.currency` row to match its joined
    `dca_machines.fiat_code`.

    Today each machine handles exactly one currency (operator-set on
    `dca_machines.fiat_code`); a deposit's currency is fully determined
    by the machine it's recorded against. The deposit dialog was
    historically a freeform text input, which let an operator typo a
    currency code (e.g., a "15 USD" row landed against an EUR Sintra
    during 2026-05-16 testing — that mismatch silently inflated the LP's
    nominal balance because the balance summary is currency-blind).

    `aiolabs/satmachineadmin#26` locks the input side; this migration
    fixes any rows already on disk. Idempotent: on a fresh install with
    no mismatches it's a no-op UPDATE.
    """
    await db.execute("""
        UPDATE spirekeeper.dca_deposits AS d
        SET currency = (
            SELECT m.fiat_code
            FROM spirekeeper.dca_machines m
            WHERE m.id = d.machine_id
        )
        WHERE EXISTS (
            SELECT 1
            FROM spirekeeper.dca_machines m
            WHERE m.id = d.machine_id
              AND m.fiat_code IS NOT NULL
              AND m.fiat_code != d.currency
        )
        """)


async def m007_add_cassette_configs(db):
    """Add cassette_configs table for operator-driven ATM cassette inventory.

    Tracks per-machine cassette state (denomination, count, position) editable
    via the spirekeeper dashboard and published to the ATM as encrypted
    kind-30078 events. See aiolabs/satmachineadmin#29 + lamassu-next#56.

    Schema choice: PK (machine_id, denomination) mirrors the ATM-side
    denomination-as-key invariant in
    bitspire/atm-tui/src/db.zig:31 and
    lamassu-next/apps/machine/electron/state-store.ts:54
    (the cassettes table PK is denomination; HAL inventory map keys on
    denomination; dispense lookup is cassetteDenominations.indexOf —
    duplicates collapse silently). Position is operator-assignable display
    order, not the addressable unit.

    Reserved nullable columns (state_count, state_at, state_event_id) hold
    the latest bitspire-cassettes-state:<atm_pubkey_hex> event the ATM
    publishes (one-shot bootstrap in v1; continuous in v2). v1 UI doesn't
    render them; v2 reconciliation UI consumes them without a migration.
    """
    await db.execute(f"""
        CREATE TABLE IF NOT EXISTS spirekeeper.cassette_configs (
            machine_id      TEXT NOT NULL,
            denomination    INTEGER NOT NULL,
            count           INTEGER NOT NULL,
            position        INTEGER NOT NULL,
            updated_at      TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
            updated_by      TEXT,
            state_count     INTEGER,
            state_at        TIMESTAMP,
            state_event_id  TEXT,
            PRIMARY KEY (machine_id, denomination)
        );
        """)


async def m008_flip_cassette_configs_pk_to_position(db):
    """Flip cassette_configs PK from (machine_id, denomination) to
    (machine_id, position). The denomination-keyed shape from m007 was
    wrong: real machines have N cartridges of the same denomination
    (cash-out throughput requires multiple bays for one denom), and the
    operator needs to swap cartridge denominations during refill ($20
    bay becomes $50 bay) without a re-provisioning event.

    Coordinated v1.1 fix with the ATM side per the 2026-05-30T18:30Z +
    18:45Z log entries:
      - Wire shape flips from {denominations: {<d>: {position, count}}}
        to {positions: {<p>: {denomination, count}}}
      - Position becomes the fixed row identity (hardware bay number);
        denomination + count are operator-editable per row
      - NO unique constraint on denomination (multiple same-denom cassettes
        are operationally valid)

    Also adds `state_denomination` nullable column reserved for v2
    reverse-channel reconciliation (operator-believed denomination per
    slot vs ATM-reported denomination — diff highlighting in v2 UI).

    SQLite doesn't support ALTER PRIMARY KEY directly; the migration
    does the standard create-copy-drop-rename dance. Idempotent via the
    column-probe trick used elsewhere in this file.
    """
    try:
        # Probe: does the old PK shape still exist? If state_denomination
        # column already exists, m008 already ran — no-op.
        await db.fetchone(
            "SELECT state_denomination FROM spirekeeper.cassette_configs " "LIMIT 1"
        )
        return
    except Exception:
        pass

    await db.execute(f"""
        CREATE TABLE IF NOT EXISTS spirekeeper.cassette_configs_new (
            machine_id          TEXT NOT NULL,
            position            INTEGER NOT NULL,
            denomination        INTEGER NOT NULL,
            count               INTEGER NOT NULL,
            updated_at          TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
            updated_by          TEXT,
            state_denomination  INTEGER,
            state_count         INTEGER,
            state_at            TIMESTAMP,
            state_event_id      TEXT,
            PRIMARY KEY (machine_id, position)
        );
        """)

    # Backfill from the old table — column-by-column copy. In the v1
    # m007 schema the row's `denomination` was simultaneously the
    # operator-believed denomination AND the ATM-reported denomination
    # (because the only write path was the bootstrap consumer copying
    # from the ATM's state.db). So state_denomination at migration time
    # = current denomination as a best-guess baseline; the next bootstrap
    # event re-populates the state_* columns authoritatively.
    await db.execute("""
        INSERT INTO spirekeeper.cassette_configs_new
            (machine_id, position, denomination, count,
             updated_at, updated_by,
             state_denomination, state_count, state_at, state_event_id)
        SELECT machine_id, position, denomination, count,
               updated_at, updated_by,
               denomination, state_count, state_at, state_event_id
        FROM spirekeeper.cassette_configs
        """)

    await db.execute("DROP TABLE spirekeeper.cassette_configs")
    await db.execute(
        "ALTER TABLE spirekeeper.cassette_configs_new " "RENAME TO cassette_configs"
    )


async def m009_split_fee_fractions_by_direction(db):
    """Split the singleton `super_fee_fraction` into per-direction fields
    and add matching per-machine operator fee fractions. Adds the
    `fee_mismatch_sats` audit column on settlements.

    Architectural intent (per aiolabs/satmachineadmin#37):
      - Super (lnbits administrator) sets X_in% and X_out% — applies
        across every machine on the lnbits instance, calculated against
        principal.
      - Operator (per-machine) sets Y_in% and Y_out% — sits on top of
        super, calculated against principal.
      - Total fee charged customer = (X+Y)% of principal per direction.
      - Distribution: super gets X% of principal; operator gets Y%
        (distributed through commission legs as today).

    Fixes the load-bearing bug where the old `super_fee_fraction` was
    interpreted as fraction-of-fee, under-paying the super by ~13× per
    cashout. The post-migration split math (bitspire.py:parse_settlement
    + calculations.py:split_principal_based) is principal-based.

    Schema delta:
      - super_config gains  super_cash_in_fee_fraction +
                            super_cash_out_fee_fraction (both backfilled
                            from the existing super_fee_fraction so live
                            config preserves intent across migrate-up).
      - dca_machines gains  operator_cash_in_fee_fraction +
                            operator_cash_out_fee_fraction (default 0;
                            operators set via the new UI surface).
      - dca_settlements gains  fee_mismatch_sats BIGINT NULL — records
                            bitspire-reported fee minus expected per
                            spirekeeper's principal-based recompute.
                            Phase 1 observability: log + record, never
                            reject (per coord-log §2026-06-01T07:00Z
                            lnbits advisory; option A locked).

    Idempotency via column-probe pattern (same shape as m006's rename
    sweep). The deprecated `super_config.super_fee_fraction` singleton
    is backfilled into the new directional fields, then dropped in the
    same migration — strict-from-the-start per workspace CLAUDE.md
    "Backwards-compatibility on pre-public-launch code" (v2-bitspire
    hasn't shipped to public users).
    """
    additions = [
        ("super_config", "super_cash_in_fee_fraction", "DECIMAL(10,4) NOT NULL DEFAULT 0.0000"),
        ("super_config", "super_cash_out_fee_fraction", "DECIMAL(10,4) NOT NULL DEFAULT 0.0000"),
        ("dca_machines", "operator_cash_in_fee_fraction", "DECIMAL(10,4) NOT NULL DEFAULT 0.0000"),
        ("dca_machines", "operator_cash_out_fee_fraction", "DECIMAL(10,4) NOT NULL DEFAULT 0.0000"),
        ("dca_settlements", "fee_mismatch_sats", "BIGINT"),
    ]
    for table, col, coltype in additions:
        try:
            await db.fetchone(f"SELECT {col} FROM spirekeeper.{table} LIMIT 1")
            # column already present — migration partially-ran previously, skip
            continue
        except Exception:
            pass
        await db.execute(
            f"ALTER TABLE spirekeeper.{table} ADD COLUMN {col} {coltype}"
        )

    # Backfill + drop the legacy singleton, gated on the column still
    # existing. Once dropped, a re-run of this migration skips both
    # steps cleanly.
    try:
        await db.fetchone(
            "SELECT super_fee_fraction FROM spirekeeper.super_config LIMIT 1"
        )
        legacy_present = True
    except Exception:
        legacy_present = False

    if legacy_present:
        # Carry the live deployment's super_fee_fraction setting forward
        # into both directional fields, but only when the operator hasn't
        # already explicitly set per-direction values (i.e., both are
        # still at DEFAULT 0).
        await db.execute(
            """
            UPDATE spirekeeper.super_config
               SET super_cash_in_fee_fraction = super_fee_fraction,
                   super_cash_out_fee_fraction = super_fee_fraction
             WHERE super_cash_in_fee_fraction = 0
               AND super_cash_out_fee_fraction = 0
               AND super_fee_fraction > 0
            """
        )
        await db.execute(
            "ALTER TABLE spirekeeper.super_config DROP COLUMN super_fee_fraction"
        )