//! ISAM: indizierter Satzzugriff auf Tabellen einer Datenbankdatei.
//!
//! Aufbau nach `openspec/changes/phase-3-isam/design.md`:
//!
//! - **D1** — die Datenbankdatei ist eine `redb`-Datei; B-Baum,
//!   Transaktionen und Crash-Sicherheit kommen von dort. Diese Datei ist
//!   der einzige Ort im Projekt, der `redb` kennt.
//! - **D2** — je Tabelle eine Satztabelle `satz-id → Satzbytes` und je
//!   Index eine Tabelle `Schlüsselbytes → satz-id`. Satz-IDs sind monoton
//!   und werden nie wiederverwendet, damit ein Cursor nach `DELETE` nie
//!   still auf einen fremden Satz zeigt.
//! - **D3/D4** — die Schlüsselkodierung ist ordnungserhaltend: der
//!   Byte-Vergleich der Kodierung entspricht dem fachlichen Vergleich.
//!   Texte ordnen nach Unicode-Codepoints — dieselbe Ordnung, die
//!   `CmpStr` der VM auf Strings anwendet (UTF-8-Bytes).
//! - **D5** — der Cursor ist ein Wert (aktiver Index, letzte Satz-ID,
//!   Kennzeichen unpositioniert), kein gehaltener Iterator. Jede Bewegung
//!   ist eine frische Bereichsabfrage.
//! - **D6** — Sicherungspunkte laufen über ein eigenes Rücknahmeprotokoll;
//!   `ROLLBACK ALL` nutzt den Abbruch der Bibliothekstransaktion direkt.
//!
//! Satzbytes entstehen mit `fileio::wert_schreiben`/`wert_lesen`, also mit
//! denselben Recordpuffern und derselben UTF-32-Festtextkodierung wie
//! `GET`/`PUT` — die Datenbankdatei ist damit aus demselben Grund nicht
//! binärkompatibel zum Vorbild wie die Binärdateien.

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};

use redb::{Database, ReadableTable, ReadableTableMetadata, TableDefinition, WriteTransaction};

use crate::errors::RuntimeError;
use crate::fileio;
use crate::value::{TypeInit, UdtLayout, Value};

// ---- Fehlercodes des ISAM-Bereichs -----------------------------------------

const NAME_UNGUELTIG: RuntimeError = RuntimeError(81);
const TABELLE_FEHLT: RuntimeError = RuntimeError(82);
const INDEX_FEHLT: RuntimeError = RuntimeError(83);
const SPALTE_UNGUELTIG: RuntimeError = RuntimeError(84);
const KEIN_SATZ: RuntimeError = RuntimeError(85);
const DOPPELTER_SCHLUESSEL: RuntimeError = RuntimeError(86);
const NULL_INDEX: RuntimeError = RuntimeError(87);
const INKONSISTENT: RuntimeError = RuntimeError(88);
const PUFFER_ZU_KLEIN: RuntimeError = RuntimeError(89);

/// Formatversion der Datenbankdatei. Eine höhere Version wird abgewiesen,
/// statt fehlinterpretiert zu werden (siehe docs/dateiformate.md).
pub const FORMATVERSION: u32 = 1;

/// Vorgabe der ISAM-Puffergrenze in Bytes (D7). Das Vorbild rechnete im
/// DOS-Speichermodell; hier ist es schlicht eine Obergrenze für die Bytes,
/// die Sätze und das Rücknahmeprotokoll gleichzeitig belegen dürfen.
pub const PUFFER_VORGABE: i64 = 65_536;

// ---- redb-Tabellen ----------------------------------------------------------

const META: &str = "tb_meta";

fn meta_def() -> TableDefinition<'static, &'static str, &'static [u8]> {
    TableDefinition::new(META)
}

fn satz_def(tabelle: &str) -> String {
    format!("satz/{tabelle}")
}

fn index_def(tabelle: &str, index: &str) -> String {
    format!("idx/{tabelle}/{index}")
}

type SatzTab<'a> = TableDefinition<'a, u64, &'static [u8]>;
type IndexTab<'a> = TableDefinition<'a, &'static [u8], u64>;

// ---- Tabellen- und Indexbeschreibung ---------------------------------------

/// Eine Spalte eines Index: Feldindex im Satztyp und Richtung.
#[derive(Debug, Clone, PartialEq, Eq)]
struct Indexspalte {
    feld: usize,
    absteigend: bool,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct Indexdef {
    name: String,
    eindeutig: bool,
    spalten: Vec<Indexspalte>,
}

/// Layout einer Tabelle: Spaltennamen mit ihren Feldtypen.
#[derive(Debug, Clone, PartialEq)]
struct Layout {
    spalten: Vec<(String, TypeInit)>,
}

impl Layout {
    /// Kurzform für den Layoutabgleich beim Öffnen einer bestehenden
    /// Tabelle. Weicht sie ab, ist die Datenbank für diesen Satztyp
    /// inkonsistent (Fehler 88).
    fn signatur(&self) -> String {
        self.spalten
            .iter()
            .map(|(n, t)| format!("{n}:{}", typ_kuerzel(t)))
            .collect::<Vec<_>>()
            .join(";")
    }

    fn spalte(&self, name: &str) -> Option<usize> {
        let gesucht = name.to_uppercase();
        self.spalten
            .iter()
            .position(|(n, _)| n.to_uppercase() == gesucht)
    }
}

fn typ_kuerzel(t: &TypeInit) -> String {
    match t {
        TypeInit::Int => "I2".into(),
        TypeInit::Lng => "I4".into(),
        TypeInit::Sng => "R4".into(),
        TypeInit::Dbl => "R8".into(),
        TypeInit::Cur => "CY".into(),
        TypeInit::FixedStr(n) => format!("T{n}"),
        TypeInit::Udt(i) => format!("U{i}"),
        TypeInit::Str => "VS".into(),
        TypeInit::Empty => "--".into(),
    }
}

// ---- Bindung einer Dateinummer ---------------------------------------------

/// Was an einer Dateinummer hängt: Tabelle, Satzlayout und der Cursor (D5).
#[derive(Debug, Clone)]
struct Bindung {
    pfad: PathBuf,
    tabelle: String,
    layout: Layout,
    udt: u16,
    /// Aktiver Index; leer = NULL-Index, also Einfügereihenfolge.
    index: String,
    /// Zuletzt besuchte Satz-ID; `None` = unpositioniert (Fehler 85).
    satz: Option<u64>,
    eof: bool,
    bof: bool,
}

// ---- Rücknahmeprotokoll (D6) -----------------------------------------------

/// Ein Protokolleintrag beschreibt, wie eine Satzoperation rückgängig
/// gemacht wird. Er entsteht bei jeder Satzoperation innerhalb einer
/// Transaktion.
#[derive(Debug, Clone)]
enum Undo {
    /// Der Satz wurde eingefügt — Rücknahme entfernt ihn wieder.
    Eingefuegt { pfad: PathBuf, tabelle: String, id: u64 },
    /// Der Satz wurde geändert oder gelöscht — Rücknahme stellt die
    /// alten Bytes wieder her.
    Vorher { pfad: PathBuf, tabelle: String, id: u64, bytes: Vec<u8> },
}

impl Undo {
    fn bytes(&self) -> usize {
        match self {
            Undo::Eingefuegt { .. } => 0,
            Undo::Vorher { bytes, .. } => bytes.len(),
        }
    }
}

/// Laufende Transaktion (`BEGINTRANS` … `COMMITTRANS`/`ROLLBACK ALL`).
struct Transaktion {
    /// Je Datenbankdatei eine Schreibtransaktion. `redb` lässt nur eine
    /// gleichzeitig zu; sie hier zu halten ist genau der Grund, warum der
    /// Cursor ein Wert und kein Iterator ist (D5).
    txns: BTreeMap<PathBuf, WriteTransaction>,
    log: Vec<Undo>,
    /// (Kennung, Position im Protokoll).
    punkte: Vec<(i16, usize)>,
    naechste_kennung: i16,
}

// ---- Zustand ---------------------------------------------------------------

/// Bewegungsrichtung der `MOVE`-Familie.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Richtung {
    Erster,
    Letzter,
    Naechster,
    Voriger,
}

/// Suchart der `SEEK`-Familie.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Suchart {
    Gleich,
    Groesser,
    GroesserGleich,
}

/// Der ISAM-Zustand eines laufenden Programms.
pub struct Isam {
    /// Offene Datenbankdateien nach normiertem Pfad. Sie bleiben bis zum
    /// Programmende offen, auch wenn alle Bindungen geschlossen sind —
    /// `redb` verträgt kein zweites Öffnen derselben Datei im Prozess.
    dbs: BTreeMap<PathBuf, Database>,
    bindungen: BTreeMap<i32, Bindung>,
    trans: Option<Transaktion>,
    /// UDT-Layouts des laufenden Moduls; die VM setzt sie beim Start.
    pub udts: Vec<UdtLayout>,
    grenze: i64,
    belegt: i64,
}

impl Default for Isam {
    fn default() -> Self {
        Isam {
            dbs: BTreeMap::new(),
            bindungen: BTreeMap::new(),
            trans: None,
            udts: Vec::new(),
            grenze: PUFFER_VORGABE,
            belegt: 0,
        }
    }
}

impl std::fmt::Debug for Isam {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Isam")
            .field("dbs", &self.dbs.keys().collect::<Vec<_>>())
            .field("bindungen", &self.bindungen.keys().collect::<Vec<_>>())
            .field("in_transaktion", &self.trans.is_some())
            .finish()
    }
}

// ---- Namensprüfung (Fehler 81) ---------------------------------------------

/// Tabellen- und Indexnamen: nicht leer, höchstens 31 Zeichen, Buchstaben,
/// Ziffern und `_`, nicht mit einer Ziffer beginnend. Der Name wird Teil
/// eines `redb`-Tabellennamens, deshalb sind `/` und `:` ausgeschlossen.
fn name_pruefen(name: &str) -> Result<(), RuntimeError> {
    let mut zeichen = name.chars();
    match zeichen.next() {
        Some(c) if c.is_alphabetic() || c == '_' => {}
        _ => return Err(NAME_UNGUELTIG),
    }
    if name.chars().count() > 31 {
        return Err(NAME_UNGUELTIG);
    }
    if !zeichen.all(|c| c.is_alphanumeric() || c == '_') {
        return Err(NAME_UNGUELTIG);
    }
    Ok(())
}

// ---- Schlüsselkodierung (D3) -----------------------------------------------

/// Kodiert einen Spaltenwert ordnungserhaltend und hängt ihn an `out`.
///
/// Zahlen liegen vorzeichenrichtig in fester Breite (Big-Endian mit
/// gekipptem Vorzeichenbit), Gleitkommazahlen zusätzlich mit invertierter
/// Mantisse bei negativem Vorzeichen. Texte liegen als UTF-8 — dessen
/// Bytereihenfolge ist die Codepoint-Reihenfolge und damit dieselbe
/// Ordnung, die der Vergleichsoperator des Dialekts anwendet (D4).
///
/// Für mehrspaltige Schlüssel braucht Text einen eindeutigen Abschluss:
/// `00` im Text wird zu `00 FF` verdoppelt, das Ende ist `00 00`. Ein
/// echtes Präfix endet damit vor jedem längeren Text.
///
/// ponytail: `NaN` ordnet oberhalb von `+inf` statt undefiniert zu sein.
/// Ceiling: Sollte der Dialekt je `NaN` in einer indizierten Spalte
/// erlauben, hier eine eigene Sonderstellung festlegen.
fn schluessel_anhaengen(out: &mut Vec<u8>, wert: &Value, typ: &TypeInit, absteigend: bool) {
    let start = out.len();
    match typ {
        TypeInit::Int => {
            let v = crate::value::as_f64(wert) as i64 as i16;
            out.extend_from_slice(&((v as u16) ^ 0x8000).to_be_bytes());
        }
        TypeInit::Lng => {
            let v = crate::value::as_f64(wert) as i64 as i32;
            out.extend_from_slice(&((v as u32) ^ 0x8000_0000).to_be_bytes());
        }
        TypeInit::Cur => {
            let v = match wert {
                Value::Cur(c) => *c,
                v => crate::value::f64_to_cur(crate::value::as_f64(v)).unwrap_or(0),
            };
            out.extend_from_slice(&((v as u64) ^ 0x8000_0000_0000_0000).to_be_bytes());
        }
        TypeInit::Sng => {
            let bits = (crate::value::as_f64(wert) as f32).to_bits();
            let k = if bits & 0x8000_0000 != 0 { !bits } else { bits | 0x8000_0000 };
            out.extend_from_slice(&k.to_be_bytes());
        }
        TypeInit::Dbl => {
            let bits = crate::value::as_f64(wert).to_bits();
            let k = if bits & 0x8000_0000_0000_0000 != 0 {
                !bits
            } else {
                bits | 0x8000_0000_0000_0000
            };
            out.extend_from_slice(&k.to_be_bytes());
        }
        // Feste Strings werden auf ihre Feldbreite gebracht, bevor sie
        // kodiert werden — sonst fände `SEEKEQ #1, "Meier"` den als
        // `"Meier   "` abgelegten Satz nicht.
        TypeInit::FixedStr(n) => {
            let s = match wert {
                Value::Str(s) => s.to_string(),
                v => crate::format::format_str_fn(v),
            };
            let mut feld: String = s.chars().take(*n as usize).collect();
            let fehlt = *n as usize - feld.chars().count();
            feld.extend(std::iter::repeat_n(' ', fehlt));
            text_anhaengen(out, &feld);
        }
        TypeInit::Str => {
            let s = match wert {
                Value::Str(s) => s.to_string(),
                v => crate::format::format_str_fn(v),
            };
            text_anhaengen(out, &s);
        }
        // Zusammengesetzte Felder haben keine fachliche Ordnung; sie
        // werden vom Aufrufer als ungültige Spalte abgewiesen.
        TypeInit::Udt(_) | TypeInit::Empty => {}
    }
    if absteigend {
        for b in &mut out[start..] {
            *b = !*b;
        }
    }
}

fn text_anhaengen(out: &mut Vec<u8>, s: &str) {
    for b in s.as_bytes() {
        out.push(*b);
        if *b == 0 {
            out.push(0xFF);
        }
    }
    out.extend_from_slice(&[0, 0]);
}

/// Kann über diesen Feldtyp geordnet werden?
fn ordenbar(t: &TypeInit) -> bool {
    !matches!(t, TypeInit::Udt(_) | TypeInit::Empty)
}

/// Kleinster Schlüssel, der echt über allen Schlüsseln mit Präfix `p`
/// liegt. `None`, wenn `p` nur aus `0xFF` besteht — dann gibt es keine
/// obere Schranke und die Bereichsabfrage läuft bis zum Ende.
fn praefix_ende(p: &[u8]) -> Option<Vec<u8>> {
    let mut v = p.to_vec();
    while let Some(letztes) = v.pop() {
        if letztes != 0xFF {
            v.push(letztes + 1);
            return Some(v);
        }
    }
    None
}

// ---- Meta-Zugriffe ----------------------------------------------------------

fn meta_lesen(txn: &WriteTransaction, schluessel: &str) -> Result<Option<Vec<u8>>, RuntimeError> {
    let tab = txn.open_table(meta_def()).map_err(|_| INKONSISTENT)?;
    let g = tab.get(schluessel).map_err(|_| INKONSISTENT)?;
    Ok(g.map(|v| v.value().to_vec()))
}

fn meta_schreiben(
    txn: &WriteTransaction,
    schluessel: &str,
    wert: &[u8],
) -> Result<(), RuntimeError> {
    let mut tab = txn.open_table(meta_def()).map_err(|_| INKONSISTENT)?;
    tab.insert(schluessel, wert).map_err(|_| INKONSISTENT)?;
    Ok(())
}

fn meta_entfernen(txn: &WriteTransaction, schluessel: &str) -> Result<(), RuntimeError> {
    let mut tab = txn.open_table(meta_def()).map_err(|_| INKONSISTENT)?;
    tab.remove(schluessel).map_err(|_| INKONSISTENT)?;
    Ok(())
}

/// Indexdefinitionen einer Tabelle: eine Zeile je Index,
/// `name\teindeutig\tfeld[,feld…]`, absteigende Spalten mit `-`.
fn indizes_lesen(txn: &WriteTransaction, tabelle: &str) -> Result<Vec<Indexdef>, RuntimeError> {
    let Some(roh) = meta_lesen(txn, &format!("idx:{tabelle}"))? else {
        return Ok(Vec::new());
    };
    let text = String::from_utf8(roh).map_err(|_| INKONSISTENT)?;
    let mut out = Vec::new();
    for zeile in text.lines().filter(|z| !z.is_empty()) {
        let teile: Vec<&str> = zeile.split('\t').collect();
        if teile.len() != 3 {
            return Err(INKONSISTENT);
        }
        let mut spalten = Vec::new();
        for s in teile[2].split(',').filter(|s| !s.is_empty()) {
            let (absteigend, zahl) = match s.strip_prefix('-') {
                Some(r) => (true, r),
                None => (false, s),
            };
            spalten.push(Indexspalte {
                feld: zahl.parse().map_err(|_| INKONSISTENT)?,
                absteigend,
            });
        }
        out.push(Indexdef {
            name: teile[0].to_string(),
            eindeutig: teile[1] == "1",
            spalten,
        });
    }
    Ok(out)
}

fn indizes_schreiben(
    txn: &WriteTransaction,
    tabelle: &str,
    defs: &[Indexdef],
) -> Result<(), RuntimeError> {
    let text = defs
        .iter()
        .map(|d| {
            let sp = d
                .spalten
                .iter()
                .map(|s| {
                    if s.absteigend {
                        format!("-{}", s.feld)
                    } else {
                        s.feld.to_string()
                    }
                })
                .collect::<Vec<_>>()
                .join(",");
            format!("{}\t{}\t{sp}", d.name, if d.eindeutig { 1 } else { 0 })
        })
        .collect::<Vec<_>>()
        .join("\n");
    meta_schreiben(txn, &format!("idx:{tabelle}"), text.as_bytes())
}

fn naechste_id(txn: &WriteTransaction, tabelle: &str) -> Result<u64, RuntimeError> {
    let schluessel = format!("seq:{tabelle}");
    let jetzt = match meta_lesen(txn, &schluessel)? {
        Some(b) if b.len() == 8 => u64::from_le_bytes(b.try_into().unwrap()),
        Some(_) => return Err(INKONSISTENT),
        None => 1,
    };
    meta_schreiben(txn, &schluessel, &(jetzt + 1).to_le_bytes())?;
    Ok(jetzt)
}

// ---- Satz- und Indexpflege --------------------------------------------------

/// Schlüsselbytes eines Satzes für einen Index, inklusive der angehängten
/// Satz-ID. Die ID macht auch bei mehrdeutigen Indizes jeden Eintrag
/// eindeutig und hält Dubletten in Einfügereihenfolge.
fn indexschluessel(def: &Indexdef, felder: &[Value], layout: &Layout, id: u64) -> Vec<u8> {
    let mut k = Vec::new();
    for s in &def.spalten {
        let leer = Value::Empty;
        let wert = felder.get(s.feld).unwrap_or(&leer);
        let typ = layout
            .spalten
            .get(s.feld)
            .map(|(_, t)| t.clone())
            .unwrap_or(TypeInit::Empty);
        schluessel_anhaengen(&mut k, wert, &typ, s.absteigend);
    }
    k.extend_from_slice(&id.to_be_bytes());
    k
}

/// Schlüsselpräfix aus den Suchwerten — ohne Satz-ID, damit es alle
/// Einträge mit diesem Präfix umfasst (Präfixsuche, D3).
fn suchschluessel(
    def: &Indexdef,
    werte: &[Value],
    layout: &Layout,
) -> Result<Vec<u8>, RuntimeError> {
    if werte.len() > def.spalten.len() {
        return Err(RuntimeError::ILLEGAL_FUNCTION_CALL);
    }
    let mut k = Vec::new();
    for (s, w) in def.spalten.iter().zip(werte) {
        let typ = layout
            .spalten
            .get(s.feld)
            .map(|(_, t)| t.clone())
            .unwrap_or(TypeInit::Empty);
        schluessel_anhaengen(&mut k, w, &typ, s.absteigend);
    }
    Ok(k)
}

impl Isam {
    // ---- Öffnen und Schließen ----------------------------------------------

    /// `OPEN datenbank$ FOR ISAM typ tabelle$ AS #n`.
    pub fn oeffnen(
        &mut self,
        nummer: i32,
        datenbank: &str,
        tabelle: &str,
        spaltennamen: &str,
        udt: u16,
    ) -> Result<(), RuntimeError> {
        if nummer < 1 {
            return Err(RuntimeError::BAD_FILE_NAME_OR_NUMBER);
        }
        if self.bindungen.contains_key(&nummer) {
            return Err(RuntimeError::FILE_ALREADY_OPEN);
        }
        if datenbank.trim().is_empty() {
            return Err(NAME_UNGUELTIG);
        }
        name_pruefen(tabelle)?;

        let felder = self
            .udts
            .get(udt as usize)
            .ok_or(RuntimeError::TYPE_MISMATCH)?
            .fields
            .clone();
        let namen: Vec<&str> = spaltennamen.split(',').filter(|s| !s.is_empty()).collect();
        if namen.len() != felder.len() {
            return Err(RuntimeError::TYPE_MISMATCH);
        }
        let layout = Layout {
            spalten: namen
                .iter()
                .map(|n| n.to_string())
                .zip(felder)
                .collect(),
        };

        let pfad = fileio::pfad_normieren(datenbank);
        self.db_oeffnen(&pfad)?;

        // Tabelle anlegen oder ihr Layout abgleichen.
        let db = self.dbs.get(&pfad).ok_or(INKONSISTENT)?;
        let txn = db.begin_write().map_err(|_| INKONSISTENT)?;
        let ergebnis = (|| -> Result<(), RuntimeError> {
            let schluessel = format!("tab:{tabelle}");
            match meta_lesen(&txn, &schluessel)? {
                Some(vorhanden) => {
                    if vorhanden != layout.signatur().as_bytes() {
                        return Err(INKONSISTENT);
                    }
                }
                None => {
                    meta_schreiben(&txn, &schluessel, layout.signatur().as_bytes())?;
                    // Satztabelle sofort anlegen, damit ein Durchlauf über
                    // eine noch leere Tabelle nicht auf einen Fehler läuft.
                    let name = satz_def(tabelle);
                    let def: SatzTab = TableDefinition::new(&name);
                    txn.open_table(def).map_err(|_| INKONSISTENT)?;
                }
            }
            Ok(())
        })();
        match ergebnis {
            Ok(()) => txn.commit().map_err(|_| INKONSISTENT)?,
            Err(e) => {
                let _ = txn.abort();
                return Err(e);
            }
        }

        self.bindungen.insert(
            nummer,
            Bindung {
                pfad,
                tabelle: tabelle.to_string(),
                layout,
                udt,
                index: String::new(),
                satz: None,
                eof: false,
                bof: false,
            },
        );
        Ok(())
    }

    /// Datenbankdatei öffnen bzw. anlegen und ihre Formatversion prüfen.
    fn db_oeffnen(&mut self, pfad: &Path) -> Result<(), RuntimeError> {
        if self.dbs.contains_key(pfad) {
            return Ok(());
        }
        let neu = !pfad.exists();
        // Jeder Öffnungsfehler ist hier eine strukturell unlesbare Datei —
        // etwa weil sie gar keine Datenbankdatei ist.
        let db = Database::create(pfad).map_err(|_| INKONSISTENT)?;
        {
            let txn = db.begin_write().map_err(|_| INKONSISTENT)?;
            let ergebnis = (|| -> Result<(), RuntimeError> {
                match meta_lesen(&txn, "version")? {
                    Some(b) if b.len() == 4 => {
                        let v = u32::from_le_bytes(b.try_into().unwrap());
                        // Eine höhere Version wird abgewiesen statt
                        // fehlinterpretiert (docs/dateiformate.md).
                        if v > FORMATVERSION {
                            return Err(INKONSISTENT);
                        }
                    }
                    Some(_) => return Err(INKONSISTENT),
                    None if neu => {
                        meta_schreiben(&txn, "version", &FORMATVERSION.to_le_bytes())?
                    }
                    // Bestehende Datei ohne Versionsmarke: keine Datenbank
                    // dieses Formats.
                    None => return Err(INKONSISTENT),
                }
                Ok(())
            })();
            match ergebnis {
                Ok(()) => txn.commit().map_err(|_| INKONSISTENT)?,
                Err(e) => {
                    let _ = txn.abort();
                    return Err(e);
                }
            }
        }
        self.dbs.insert(pfad.to_path_buf(), db);
        Ok(())
    }

    pub fn ist_isam(&self, nummer: i32) -> bool {
        self.bindungen.contains_key(&nummer)
    }

    /// `CLOSE #n` — Bindung lösen.
    ///
    /// Eine laufende Transaktion bleibt davon unberührt: über ihr Ende
    /// entscheiden allein `COMMITTRANS` und `ROLLBACK ALL`. Ausstehende
    /// Änderungen sind ohne Transaktion ohnehin schon festgeschrieben, weil
    /// jede Operation für sich eine Transaktion ist; würde `CLOSE` dagegen
    /// festschreiben, beendete das Schließen *irgendeiner* Dateinummer die
    /// Transaktion aller anderen mit.
    pub fn schliessen(&mut self, nummer: i32) -> Result<(), RuntimeError> {
        self.bindungen.remove(&nummer);
        Ok(())
    }

    /// Programmende bzw. `RESET`: Bindungen lösen und Dateien freigeben.
    /// Eine nicht festgeschriebene Transaktion verfällt — sie abzubrechen
    /// ist die Bedeutung von „nicht festgeschrieben".
    pub fn alles_schliessen(&mut self) {
        if let Some(t) = self.trans.take() {
            for (_, txn) in t.txns {
                let _ = txn.abort();
            }
        }
        self.belegt = 0;
        self.bindungen.clear();
        self.dbs.clear();
    }

    // ---- Transaktionsrahmen -------------------------------------------------

    /// Führt `f` in einer Schreibtransaktion aus und schreibt fest. Läuft
    /// gerade eine Transaktion (`BEGINTRANS`), nutzt `f` sie und die
    /// Festschreibung bleibt `COMMITTRANS` überlassen.
    fn schreiben<T>(
        &mut self,
        pfad: &Path,
        f: impl FnOnce(&WriteTransaction, &mut Vec<Undo>) -> Result<T, RuntimeError>,
    ) -> Result<T, RuntimeError> {
        let Isam { dbs, trans, .. } = self;
        let db = dbs.get(pfad).ok_or(INKONSISTENT)?;
        match trans {
            Some(t) => {
                if !t.txns.contains_key(pfad) {
                    let txn = db.begin_write().map_err(|_| INKONSISTENT)?;
                    t.txns.insert(pfad.to_path_buf(), txn);
                }
                let Transaktion { txns, log, .. } = t;
                let txn = txns.get(pfad).ok_or(INKONSISTENT)?;
                f(txn, log)
            }
            None => {
                let txn = db.begin_write().map_err(|_| INKONSISTENT)?;
                // Außerhalb einer Transaktion gibt es nichts zurückzunehmen;
                // das Protokoll dieser einen Operation verfällt sofort.
                let mut weg = Vec::new();
                match f(&txn, &mut weg) {
                    Ok(v) => {
                        txn.commit().map_err(|_| INKONSISTENT)?;
                        Ok(v)
                    }
                    Err(e) => {
                        let _ = txn.abort();
                        Err(e)
                    }
                }
            }
        }
    }

    /// Wie `schreiben`, aber für Operationen ohne Änderung: die eigens
    /// geöffnete Transaktion wird abgebrochen statt festgeschrieben. Es ist
    /// trotzdem eine Schreibtransaktion, weil sie sonst die noch nicht
    /// festgeschriebenen Änderungen einer laufenden Transaktion nicht sähe.
    fn lesen<T>(
        &mut self,
        pfad: &Path,
        f: impl FnOnce(&WriteTransaction) -> Result<T, RuntimeError>,
    ) -> Result<T, RuntimeError> {
        let Isam { dbs, trans, .. } = self;
        let db = dbs.get(pfad).ok_or(INKONSISTENT)?;
        match trans {
            Some(t) => {
                if !t.txns.contains_key(pfad) {
                    let txn = db.begin_write().map_err(|_| INKONSISTENT)?;
                    t.txns.insert(pfad.to_path_buf(), txn);
                }
                f(t.txns.get(pfad).ok_or(INKONSISTENT)?)
            }
            None => {
                let txn = db.begin_write().map_err(|_| INKONSISTENT)?;
                let r = f(&txn);
                let _ = txn.abort();
                r
            }
        }
    }

    fn bindung(&self, nummer: i32) -> Result<&Bindung, RuntimeError> {
        self.bindungen
            .get(&nummer)
            .ok_or(RuntimeError::BAD_FILE_NAME_OR_NUMBER)
    }

    // ---- Indexverwaltung ----------------------------------------------------

    /// `CREATEINDEX [#]n, name$, eindeutig%, spalte$[, spalte$…]`.
    /// Ein `-` vor dem Spaltennamen ordnet absteigend (Erweiterung über die
    /// Original-Hilfe hinaus, siehe umfang-und-signaturen.md des Changes).
    pub fn index_anlegen(
        &mut self,
        nummer: i32,
        name: &str,
        eindeutig: bool,
        spalten: &[String],
    ) -> Result<(), RuntimeError> {
        name_pruefen(name)?;
        let b = self.bindung(nummer)?.clone();
        if spalten.is_empty() {
            return Err(SPALTE_UNGUELTIG);
        }
        let mut def = Indexdef { name: name.to_string(), eindeutig, spalten: Vec::new() };
        for s in spalten {
            let (absteigend, sname) = match s.strip_prefix('-') {
                Some(r) => (true, r),
                None => (false, s.as_str()),
            };
            let feld = b.layout.spalte(sname).ok_or(SPALTE_UNGUELTIG)?;
            if !ordenbar(&b.layout.spalten[feld].1) {
                return Err(SPALTE_UNGUELTIG);
            }
            def.spalten.push(Indexspalte { feld, absteigend });
        }

        let layout = b.layout.clone();
        let tabelle = b.tabelle.clone();
        let udts = self.udts.clone();
        self.schreiben(&b.pfad.clone(), move |txn, _log| {
            let mut defs = indizes_lesen(txn, &tabelle)?;
            if defs.iter().any(|d| d.name.eq_ignore_ascii_case(name)) {
                return Err(RuntimeError::DUPLICATE_DEFINITION);
            }
            // Index über den vorhandenen Bestand aufbauen.
            let satz_name = satz_def(&tabelle);
            let satz_tab: SatzTab = TableDefinition::new(&satz_name);
            let saetze: Vec<(u64, Vec<u8>)> = {
                let t = txn.open_table(satz_tab).map_err(|_| INKONSISTENT)?;
                let mut v = Vec::new();
                for e in t.iter().map_err(|_| INKONSISTENT)? {
                    let (k, w) = e.map_err(|_| INKONSISTENT)?;
                    v.push((k.value(), w.value().to_vec()));
                }
                v
            };
            // Erst alle Schlüssel bilden und die Eindeutigkeit prüfen, dann
            // schreiben: sonst bliebe bei Fehler 86 innerhalb einer
            // Transaktion eine halb gefüllte Indextabelle zurück, die keine
            // Indexdefinition mehr nennt.
            let mut eintraege: Vec<(Vec<u8>, u64)> = Vec::with_capacity(saetze.len());
            for (id, bytes) in &saetze {
                let felder = felder_lesen(bytes, &layout, &udts)?;
                eintraege.push((indexschluessel(&def, &felder, &layout, *id), *id));
            }
            if def.eindeutig {
                // Die Schlüssel ohne die angehängte Satz-ID müssen paarweise
                // verschieden sein; sortiert genügt der Blick auf den Nachbarn.
                let mut ohne_id: Vec<&[u8]> =
                    eintraege.iter().map(|(k, _)| &k[..k.len() - 8]).collect();
                ohne_id.sort_unstable();
                if ohne_id.windows(2).any(|p| p[0] == p[1]) {
                    return Err(DOPPELTER_SCHLUESSEL);
                }
            }

            let idx_name = index_def(&tabelle, name);
            let idx_tab: IndexTab = TableDefinition::new(&idx_name);
            let mut t = txn.open_table(idx_tab).map_err(|_| INKONSISTENT)?;
            for (k, id) in &eintraege {
                t.insert(&k[..], *id).map_err(|_| INKONSISTENT)?;
            }
            drop(t);
            defs.push(def);
            indizes_schreiben(txn, &tabelle, &defs)
        })
    }

    /// `DELETEINDEX [#]n, name$`.
    pub fn index_loeschen(&mut self, nummer: i32, name: &str) -> Result<(), RuntimeError> {
        let b = self.bindung(nummer)?.clone();
        let tabelle = b.tabelle.clone();
        self.schreiben(&b.pfad.clone(), |txn, _| {
            let mut defs = indizes_lesen(txn, &tabelle)?;
            let Some(i) = defs.iter().position(|d| d.name.eq_ignore_ascii_case(name)) else {
                return Err(INDEX_FEHLT);
            };
            let idx_name = index_def(&tabelle, &defs[i].name);
            let idx_tab: IndexTab = TableDefinition::new(&idx_name);
            txn.delete_table(idx_tab).map_err(|_| INKONSISTENT)?;
            defs.remove(i);
            indizes_schreiben(txn, &tabelle, &defs)
        })?;
        // Ein gelöschter Index darf nicht aktiv bleiben.
        if let Some(bind) = self.bindungen.get_mut(&nummer) {
            if bind.index.eq_ignore_ascii_case(name) {
                bind.index.clear();
                bind.satz = None;
            }
        }
        Ok(())
    }

    /// `SETINDEX [#]n[, name$]` — ohne Namen (oder mit `""`) gilt der
    /// NULL-Index, also die Einfügereihenfolge. Der aktuelle Satz ist
    /// danach der erste Satz dieser Ordnung (Original-Hilfe: „After
    /// SETINDEX, the current record is the first record according to that
    /// index").
    pub fn index_setzen(&mut self, nummer: i32, name: &str) -> Result<(), RuntimeError> {
        let b = self.bindung(nummer)?.clone();
        let tabelle = b.tabelle.clone();
        let treffer = if name.is_empty() {
            String::new()
        } else {
            self.lesen(&b.pfad.clone(), |txn| {
                indizes_lesen(txn, &tabelle)?
                    .into_iter()
                    .find(|d| d.name.eq_ignore_ascii_case(name))
                    .map(|d| d.name)
                    .ok_or(INDEX_FEHLT)
            })?
        };
        let bind = self.bindungen.get_mut(&nummer).ok_or(INKONSISTENT)?;
        bind.index = treffer;
        bind.satz = None;
        bind.eof = false;
        bind.bof = false;
        // Auf den ersten Satz der neuen Ordnung stellen; bei leerer Tabelle
        // setzt das EOF und BOF.
        self.bewegen(nummer, Richtung::Erster)
    }

    /// `GETINDEX$(n)` — Name des aktiven Index, leer beim NULL-Index.
    pub fn index_name(&self, nummer: i32) -> Result<String, RuntimeError> {
        Ok(self.bindung(nummer)?.index.clone())
    }

    /// `DELETETABLE datenbank$, tabelle$` — Tabelle samt Indizes entfernen.
    pub fn tabelle_loeschen(&mut self, datenbank: &str, tabelle: &str) -> Result<(), RuntimeError> {
        if datenbank.trim().is_empty() {
            return Err(NAME_UNGUELTIG);
        }
        name_pruefen(tabelle)?;
        let pfad = fileio::pfad_normieren(datenbank);
        if !pfad.exists() {
            return Err(TABELLE_FEHLT);
        }
        self.db_oeffnen(&pfad)?;
        let tab = tabelle.to_string();
        self.schreiben(&pfad, move |txn, _| {
            if meta_lesen(txn, &format!("tab:{tab}"))?.is_none() {
                return Err(TABELLE_FEHLT);
            }
            for d in indizes_lesen(txn, &tab)? {
                let name = index_def(&tab, &d.name);
                let idx: IndexTab = TableDefinition::new(&name);
                txn.delete_table(idx).map_err(|_| INKONSISTENT)?;
            }
            let name = satz_def(&tab);
            let satz: SatzTab = TableDefinition::new(&name);
            txn.delete_table(satz).map_err(|_| INKONSISTENT)?;
            meta_entfernen(txn, &format!("tab:{tab}"))?;
            meta_entfernen(txn, &format!("idx:{tab}"))?;
            meta_entfernen(txn, &format!("seq:{tab}"))?;
            Ok(())
        })
    }

    // ---- Satzoperationen ----------------------------------------------------

    /// `INSERT [#]n, satz`.
    pub fn einfuegen(&mut self, nummer: i32, wert: &Value) -> Result<(), RuntimeError> {
        let b = self.bindung(nummer)?.clone();
        let bytes = self.satz_bytes(&b, wert)?;
        self.puffer_pruefen(bytes.len())?;
        let layout = b.layout.clone();
        let tabelle = b.tabelle.clone();
        let pfad = b.pfad.clone();
        let fuer_log = pfad.clone();
        let udts = self.udts.clone();
        let id = self.schreiben(&pfad, move |txn, log| {
            let id = naechste_id(txn, &tabelle)?;
            satz_schreiben(txn, &tabelle, &layout, &udts, id, Some(&bytes), None)?;
            log.push(Undo::Eingefuegt { pfad: fuer_log, tabelle: tabelle.clone(), id });
            Ok(id)
        })?;
        self.belegt += bytes_von(&self.trans);
        let bind = self.bindungen.get_mut(&nummer).ok_or(INKONSISTENT)?;
        // Das Vorbild lässt den Cursor beim Einfügen stehen; er zeigt auf
        // den neuen Satz, damit `INSERT` und `RETRIEVE` zusammenpassen.
        bind.satz = Some(id);
        bind.eof = false;
        bind.bof = false;
        Ok(())
    }

    /// `RETRIEVE [#]n, satz`.
    pub fn satz_lesen(&mut self, nummer: i32) -> Result<Value, RuntimeError> {
        let b = self.bindung(nummer)?.clone();
        let id = b.satz.ok_or(KEIN_SATZ)?;
        let tabelle = b.tabelle.clone();
        let udt = b.udt;
        let udts = self.udts.clone();
        self.lesen(&b.pfad.clone(), move |txn| {
            let bytes = satz_bytes_lesen(txn, &tabelle, id)?.ok_or(KEIN_SATZ)?;
            let (v, _) = fileio::wert_lesen(&bytes, 0, &TypeInit::Udt(udt), &udts)?;
            Ok(v)
        })
    }

    /// `UPDATE [#]n, satz`.
    pub fn aendern(&mut self, nummer: i32, wert: &Value) -> Result<(), RuntimeError> {
        let b = self.bindung(nummer)?.clone();
        let id = b.satz.ok_or(KEIN_SATZ)?;
        let bytes = self.satz_bytes(&b, wert)?;
        self.puffer_pruefen(bytes.len())?;
        let layout = b.layout.clone();
        let tabelle = b.tabelle.clone();
        let pfad = b.pfad.clone();
        let fuer_log = pfad.clone();
        let udts = self.udts.clone();
        self.schreiben(&pfad, move |txn, log| {
            let alt = satz_bytes_lesen(txn, &tabelle, id)?.ok_or(KEIN_SATZ)?;
            satz_schreiben(txn, &tabelle, &layout, &udts, id, Some(&bytes), Some(&alt))?;
            log.push(Undo::Vorher {
                pfad: fuer_log,
                tabelle: tabelle.clone(),
                id,
                bytes: alt,
            });
            Ok(())
        })?;
        self.belegt += bytes_von(&self.trans);
        Ok(())
    }

    /// `DELETE [#]n`.
    ///
    /// Der auf den gelöschten folgende Satz wird der aktuelle (Original-Hilfe:
    /// „the record following the deleted record becomes the current record").
    /// War der gelöschte der letzte, steht der Cursor am Ende der Tabelle und
    /// es gibt keinen aktuellen Satz.
    pub fn satz_loeschen(&mut self, nummer: i32) -> Result<(), RuntimeError> {
        let b = self.bindung(nummer)?.clone();
        let id = b.satz.ok_or(KEIN_SATZ)?;
        // Der Nachfolger muss vor dem Löschen bestimmt werden — danach ist
        // der Schlüssel des aktuellen Satzes fort und die Ordnung verloren.
        let nachfolger = self.ziel_in_ordnung(nummer, Richtung::Naechster)?;
        let layout = b.layout.clone();
        let tabelle = b.tabelle.clone();
        let pfad = b.pfad.clone();
        let fuer_log = pfad.clone();
        let udts = self.udts.clone();
        self.schreiben(&pfad, move |txn, log| {
            let alt = satz_bytes_lesen(txn, &tabelle, id)?.ok_or(KEIN_SATZ)?;
            satz_schreiben(txn, &tabelle, &layout, &udts, id, None, Some(&alt))?;
            log.push(Undo::Vorher {
                pfad: fuer_log,
                tabelle: tabelle.clone(),
                id,
                bytes: alt,
            });
            Ok(())
        })?;
        self.belegt += bytes_von(&self.trans);
        let bind = self.bindungen.get_mut(&nummer).ok_or(INKONSISTENT)?;
        bind.satz = nachfolger;
        bind.eof = nachfolger.is_none();
        bind.bof = false;
        Ok(())
    }

    /// Satz einer Variablen in Bytes umwandeln — dieselbe Kodierung wie
    /// bei `PUT` auf eine Datei mit `RANDOM`-Zugriff.
    fn satz_bytes(&self, b: &Bindung, wert: &Value) -> Result<Vec<u8>, RuntimeError> {
        let typ = TypeInit::Udt(b.udt);
        let n = fileio::breite(&typ, &self.udts).ok_or(RuntimeError::TYPE_MISMATCH)?;
        let mut puffer = vec![0u8; n];
        fileio::wert_schreiben(&mut puffer, 0, wert, &typ, &self.udts)?;
        Ok(puffer)
    }

    // ---- Cursor -------------------------------------------------------------

    /// Wohin führt eine Bewegung, ohne den Cursor schon zu setzen?
    /// `DELETE` braucht dieselbe Rechnung, um den Nachfolger noch vor dem
    /// Löschen zu kennen.
    fn ziel_in_ordnung(
        &mut self,
        nummer: i32,
        richtung: Richtung,
    ) -> Result<Option<u64>, RuntimeError> {
        let b = self.bindung(nummer)?.clone();
        let layout = b.layout.clone();
        let tabelle = b.tabelle.clone();
        let index = b.index.clone();
        let satz = b.satz;
        let udts = self.udts.clone();
        self.lesen(&b.pfad.clone(), move |txn| {
            if index.is_empty() {
                return null_index_bewegen(txn, &tabelle, satz, richtung);
            }
            let def = indizes_lesen(txn, &tabelle)?
                .into_iter()
                .find(|d| d.name.eq_ignore_ascii_case(&index))
                .ok_or(INDEX_FEHLT)?;
            index_bewegen(txn, &tabelle, &def, &layout, &udts, satz, richtung)
        })
    }

    /// `MOVEFIRST`/`MOVELAST`/`MOVENEXT`/`MOVEPREVIOUS` in der Ordnung des
    /// aktiven Index; ohne aktiven Index gilt die Einfügereihenfolge.
    pub fn bewegen(&mut self, nummer: i32, richtung: Richtung) -> Result<(), RuntimeError> {
        let ziel = self.ziel_in_ordnung(nummer, richtung)?;
        let bind = self.bindungen.get_mut(&nummer).ok_or(INKONSISTENT)?;
        match ziel {
            Some(id) => {
                bind.satz = Some(id);
                bind.eof = false;
                bind.bof = false;
            }
            // Über das Ende bzw. den Anfang hinaus: Cursor unpositioniert.
            None => {
                bind.satz = None;
                let rueckwaerts = matches!(richtung, Richtung::Voriger);
                bind.eof = !rueckwaerts;
                bind.bof = rueckwaerts;
                if matches!(richtung, Richtung::Erster | Richtung::Letzter) {
                    // Leere Tabelle: beide Ränder gelten.
                    bind.eof = true;
                    bind.bof = true;
                }
            }
        }
        Ok(())
    }

    /// `SEEKEQ`/`SEEKGT`/`SEEKGE`.
    ///
    /// Über den NULL-Index lässt sich nicht suchen — er hat keine
    /// Schlüssel, sondern nur die Einfügereihenfolge. Das ist genau die
    /// Lage, die Fehler 87 im Katalog des Vorbilds benennt („ISAM -
    /// Invalid operation on NULL index"). Fehler 83 bleibt dem wirklich
    /// unbekannten Indexnamen vorbehalten (`SETINDEX`, `DELETEINDEX`).
    pub fn suchen(
        &mut self,
        nummer: i32,
        art: Suchart,
        werte: &[Value],
    ) -> Result<(), RuntimeError> {
        let b = self.bindung(nummer)?.clone();
        if b.index.is_empty() {
            return Err(NULL_INDEX);
        }
        let layout = b.layout.clone();
        let tabelle = b.tabelle.clone();
        let index = b.index.clone();
        let werte = werte.to_vec();
        let ziel = self.lesen(&b.pfad.clone(), move |txn| {
            let def = indizes_lesen(txn, &tabelle)?
                .into_iter()
                .find(|d| d.name.eq_ignore_ascii_case(&index))
                .ok_or(INDEX_FEHLT)?;
            // SEEKEQ mit unvollständigem Schlüssel schlägt immer fehl —
            // wortgetreu aus der Original-Hilfe.
            if art == Suchart::Gleich && werte.len() != def.spalten.len() {
                return Ok(None);
            }
            let praefix = suchschluessel(&def, &werte, &layout)?;
            let name = index_def(&tabelle, &def.name);
            let idx: IndexTab = TableDefinition::new(&name);
            let t = txn.open_table(idx).map_err(|_| INKONSISTENT)?;
            let vollstaendig = werte.len() == def.spalten.len();
            // `SEEKGT` überspringt die Präfixgruppe nur bei vollständigem
            // Schlüssel; mit Teilschlüssel positioniert es wie `SEEKGE`
            // auf dem ersten passenden Satz (Original-Hilfe).
            let von = if art == Suchart::Groesser && vollstaendig {
                match praefix_ende(&praefix) {
                    Some(v) => v,
                    None => return Ok(None),
                }
            } else {
                praefix.clone()
            };
            let mut bereich = t.range(von.as_slice()..).map_err(|_| INKONSISTENT)?;
            let Some(e) = bereich.next() else { return Ok(None) };
            let (k, v) = e.map_err(|_| INKONSISTENT)?;
            if art == Suchart::Gleich && !k.value().starts_with(&praefix) {
                return Ok(None);
            }
            Ok(Some(v.value()))
        })?;
        let bind = self.bindungen.get_mut(&nummer).ok_or(INKONSISTENT)?;
        match ziel {
            Some(id) => {
                bind.satz = Some(id);
                bind.eof = false;
                bind.bof = false;
            }
            // Erfolglose Suche: Ende der Tabelle, kein aktueller Satz.
            None => {
                bind.satz = None;
                bind.eof = true;
                bind.bof = false;
            }
        }
        Ok(())
    }

    pub fn eof(&self, nummer: i32) -> bool {
        self.bindungen.get(&nummer).is_some_and(|b| b.eof)
    }

    pub fn bof(&self, nummer: i32) -> bool {
        self.bindungen.get(&nummer).is_some_and(|b| b.bof)
    }

    /// `LOF(n)` für eine ISAM-Dateinummer: Zahl der Sätze in der Tabelle.
    pub fn satzzahl(&mut self, nummer: i32) -> Result<i64, RuntimeError> {
        let b = self.bindung(nummer)?.clone();
        let tabelle = b.tabelle.clone();
        self.lesen(&b.pfad.clone(), move |txn| {
            let name = satz_def(&tabelle);
            let def: SatzTab = TableDefinition::new(&name);
            let t = txn.open_table(def).map_err(|_| INKONSISTENT)?;
            Ok(t.len().map_err(|_| INKONSISTENT)? as i64)
        })
    }

    /// `LOC(n)` für eine ISAM-Dateinummer: ID des aktuellen Satzes,
    /// `0` wenn der Cursor unpositioniert ist.
    pub fn satznummer(&self, nummer: i32) -> Result<i64, RuntimeError> {
        Ok(self.bindung(nummer)?.satz.unwrap_or(0) as i64)
    }

    // ---- Transaktionen ------------------------------------------------------

    /// `BEGINTRANS`.
    pub fn trans_beginn(&mut self) -> Result<(), RuntimeError> {
        if self.trans.is_some() {
            // Verschachtelte Transaktionen kennt das Vorbild nicht; dafür
            // gibt es Sicherungspunkte.
            return Err(RuntimeError::ILLEGAL_FUNCTION_CALL);
        }
        self.trans = Some(Transaktion {
            txns: BTreeMap::new(),
            log: Vec::new(),
            punkte: Vec::new(),
            naechste_kennung: 1,
        });
        self.belegt = 0;
        Ok(())
    }

    /// `COMMITTRANS`.
    pub fn trans_festschreiben(&mut self) -> Result<(), RuntimeError> {
        let Some(t) = self.trans.take() else {
            return Ok(());
        };
        self.belegt = 0;
        for (_, txn) in t.txns {
            txn.commit().map_err(|_| INKONSISTENT)?;
        }
        Ok(())
    }

    /// `SAVEPOINT` — liefert die Kennung des gesetzten Sicherungspunkts.
    pub fn sicherungspunkt(&mut self) -> Result<i16, RuntimeError> {
        let t = self.trans.as_mut().ok_or(RuntimeError::ILLEGAL_FUNCTION_CALL)?;
        let kennung = t.naechste_kennung;
        t.naechste_kennung = t.naechste_kennung.saturating_add(1);
        let pos = t.log.len();
        t.punkte.push((kennung, pos));
        Ok(kennung)
    }

    /// `ROLLBACK`, `ROLLBACK kennung` und `ROLLBACK ALL`.
    ///
    /// `ALL` (Kennung `crate::isam::ROLLBACK_ALL`) bricht die Transaktion
    /// der Bibliothek ab — damit sind alle Änderungen seit `BEGINTRANS`
    /// in einem Schritt zurückgenommen und die Transaktion ist beendet.
    /// Ein Sicherungspunkt spielt dagegen nur das Protokoll rückwärts ab
    /// und lässt die Transaktion offen (D6).
    pub fn ruecknahme(&mut self, kennung: i32) -> Result<(), RuntimeError> {
        if self.trans.is_none() {
            return Err(RuntimeError::ILLEGAL_FUNCTION_CALL);
        }
        if kennung == ROLLBACK_ALL {
            let t = self.trans.take().expect("gerade geprüft");
            for (_, txn) in t.txns {
                let _ = txn.abort();
            }
            self.belegt = 0;
            self.cursor_loesen();
            return Ok(());
        }

        // Zielposition im Protokoll bestimmen.
        let ziel = {
            let t = self.trans.as_mut().expect("gerade geprüft");
            if kennung == 0 {
                // Bare `ROLLBACK`: bis zum letzten Sicherungspunkt, sonst
                // bis zum Transaktionsbeginn.
                t.punkte.pop().map(|(_, p)| p).unwrap_or(0)
            } else {
                let k = kennung as i16;
                let Some(i) = t.punkte.iter().position(|(kk, _)| *kk == k) else {
                    return Err(RuntimeError::ILLEGAL_FUNCTION_CALL);
                };
                let p = t.punkte[i].1;
                t.punkte.truncate(i);
                p
            }
        };
        self.protokoll_abspielen(ziel)?;
        self.cursor_loesen();
        Ok(())
    }

    /// Protokolleinträge ab `ziel` rückwärts anwenden und verwerfen.
    fn protokoll_abspielen(&mut self, ziel: usize) -> Result<(), RuntimeError> {
        loop {
            let eintrag = {
                let t = self.trans.as_mut().ok_or(INKONSISTENT)?;
                if t.log.len() <= ziel {
                    break;
                }
                t.log.pop().ok_or(INKONSISTENT)?
            };
            self.belegt -= eintrag.bytes() as i64;
            let udts = self.udts.clone();
            match eintrag {
                Undo::Eingefuegt { pfad, tabelle, id } => {
                    let layout = self.layout_von(&pfad, &tabelle)?;
                    self.schreiben(&pfad, move |txn, _| {
                        let alt = satz_bytes_lesen(txn, &tabelle, id)?;
                        satz_schreiben(
                            txn,
                            &tabelle,
                            &layout,
                            &udts,
                            id,
                            None,
                            alt.as_deref(),
                        )
                    })?;
                }
                Undo::Vorher { pfad, tabelle, id, bytes } => {
                    let layout = self.layout_von(&pfad, &tabelle)?;
                    self.schreiben(&pfad, move |txn, _| {
                        let alt = satz_bytes_lesen(txn, &tabelle, id)?;
                        satz_schreiben(
                            txn,
                            &tabelle,
                            &layout,
                            &udts,
                            id,
                            Some(&bytes),
                            alt.as_deref(),
                        )
                    })?;
                }
            }
        }
        Ok(())
    }

    /// Layout einer Tabelle über eine bestehende Bindung finden. Die
    /// Rücknahme läuft nur über Tabellen, die im Programm offen sind.
    fn layout_von(&self, pfad: &Path, tabelle: &str) -> Result<Layout, RuntimeError> {
        self.bindungen
            .values()
            .find(|b| b.pfad == pfad && b.tabelle == tabelle)
            .map(|b| b.layout.clone())
            .ok_or(INKONSISTENT)
    }

    /// Nach jeder Rücknahme gilt der Cursor als unpositioniert.
    fn cursor_loesen(&mut self) {
        for b in self.bindungen.values_mut() {
            b.satz = None;
            b.eof = false;
            b.bof = false;
        }
    }

    // ---- Pufferverwaltung (D7) ---------------------------------------------

    /// `SETMEM(n)` — Grenze um `n` Bytes ändern, verfügbare Größe liefern.
    pub fn setmem(&mut self, delta: i64) -> i64 {
        self.grenze = (self.grenze + delta).max(0);
        (self.grenze - self.belegt).max(0)
    }

    fn puffer_pruefen(&self, benoetigt: usize) -> Result<(), RuntimeError> {
        if self.belegt + benoetigt as i64 > self.grenze {
            return Err(PUFFER_ZU_KLEIN);
        }
        Ok(())
    }
}

/// Sentinel-Kennung für `ROLLBACK ALL` (Gegenstück zu `ast::ROLLBACK_ALL`).
pub const ROLLBACK_ALL: i32 = -1;

/// Wieviel Puffer belegt der zuletzt angelegte Protokolleintrag? Außerhalb
/// einer Transaktion verfällt er sofort und belegt nichts.
fn bytes_von(trans: &Option<Transaktion>) -> i64 {
    trans
        .as_ref()
        .and_then(|t| t.log.last())
        .map(|e| e.bytes() as i64)
        .unwrap_or(0)
}

// ---- Satz- und Indexpflege (freie Funktionen auf einer Transaktion) --------

fn satz_bytes_lesen(
    txn: &WriteTransaction,
    tabelle: &str,
    id: u64,
) -> Result<Option<Vec<u8>>, RuntimeError> {
    let name = satz_def(tabelle);
    let def: SatzTab = TableDefinition::new(&name);
    let t = txn.open_table(def).map_err(|_| INKONSISTENT)?;
    let gefunden = t.get(id).map_err(|_| INKONSISTENT)?.map(|v| v.value().to_vec());
    Ok(gefunden)
}

/// Feldwerte eines Satzes aus seinen Bytes.
fn felder_lesen(
    bytes: &[u8],
    layout: &Layout,
    udts: &[UdtLayout],
) -> Result<Vec<Value>, RuntimeError> {
    let mut off = 0;
    let mut out = Vec::with_capacity(layout.spalten.len());
    for (_, t) in &layout.spalten {
        let (v, n) = fileio::wert_lesen(bytes, off, t, udts)?;
        out.push(v);
        off += n;
    }
    Ok(out)
}

/// Satz schreiben, löschen oder ersetzen und dabei **alle** Indizes
/// nachführen. `neu = None` löscht, `alt = None` fügt ein.
///
/// Die Eindeutigkeitsprüfung läuft vor jeder Änderung: schlägt sie fehl,
/// ist noch nichts geschrieben und der Satzbestand bleibt unverändert.
fn satz_schreiben(
    txn: &WriteTransaction,
    tabelle: &str,
    layout: &Layout,
    udts: &[UdtLayout],
    id: u64,
    neu: Option<&[u8]>,
    alt: Option<&[u8]>,
) -> Result<(), RuntimeError> {
    let defs = indizes_lesen(txn, tabelle)?;
    let neue_felder = match neu {
        Some(b) => Some(felder_lesen(b, layout, udts)?),
        None => None,
    };
    let alte_felder = match alt {
        Some(b) => Some(felder_lesen(b, layout, udts)?),
        None => None,
    };

    // Eindeutigkeit zuerst prüfen — vor jeder Schreiboperation.
    if let Some(felder) = &neue_felder {
        for d in defs.iter().filter(|d| d.eindeutig) {
            let k = indexschluessel(d, felder, layout, id);
            let praefix = &k[..k.len() - 8];
            let name = index_def(tabelle, &d.name);
            let idx: IndexTab = TableDefinition::new(&name);
            let t = txn.open_table(idx).map_err(|_| INKONSISTENT)?;
            let kollision = {
                let mut bereich = t.range(praefix..).map_err(|_| INKONSISTENT)?;
                match bereich.next() {
                    Some(e) => {
                        let (kk, vv) = e.map_err(|_| INKONSISTENT)?;
                        kk.value().starts_with(praefix) && vv.value() != id
                    }
                    None => false,
                }
            };
            if kollision {
                return Err(DOPPELTER_SCHLUESSEL);
            }
        }
    }

    // Alte Indexeinträge entfernen, neue setzen.
    for d in &defs {
        let name = index_def(tabelle, &d.name);
        let idx: IndexTab = TableDefinition::new(&name);
        let mut t = txn.open_table(idx).map_err(|_| INKONSISTENT)?;
        if let Some(felder) = &alte_felder {
            let k = indexschluessel(d, felder, layout, id);
            t.remove(&k[..]).map_err(|_| INKONSISTENT)?;
        }
        if let Some(felder) = &neue_felder {
            let k = indexschluessel(d, felder, layout, id);
            t.insert(&k[..], id).map_err(|_| INKONSISTENT)?;
        }
    }

    let name = satz_def(tabelle);
    let def: SatzTab = TableDefinition::new(&name);
    let mut t = txn.open_table(def).map_err(|_| INKONSISTENT)?;
    match neu {
        Some(b) => {
            t.insert(id, b).map_err(|_| INKONSISTENT)?;
        }
        None => {
            t.remove(id).map_err(|_| INKONSISTENT)?;
        }
    }
    Ok(())
}

/// Bewegung ohne aktiven Index: Einfügereihenfolge, also die Ordnung der
/// monoton vergebenen Satz-IDs.
fn null_index_bewegen(
    txn: &WriteTransaction,
    tabelle: &str,
    satz: Option<u64>,
    richtung: Richtung,
) -> Result<Option<u64>, RuntimeError> {
    let name = satz_def(tabelle);
    let def: SatzTab = TableDefinition::new(&name);
    let t = txn.open_table(def).map_err(|_| INKONSISTENT)?;
    let id = match richtung {
        Richtung::Erster => t.first().map_err(|_| INKONSISTENT)?.map(|(k, _)| k.value()),
        Richtung::Letzter => t.last().map_err(|_| INKONSISTENT)?.map(|(k, _)| k.value()),
        Richtung::Naechster => {
            // Ohne aktuellen Satz beginnt `MOVENEXT` am Anfang.
            let Some(s) = satz else {
                return Ok(t.first().map_err(|_| INKONSISTENT)?.map(|(k, _)| k.value()));
            };
            let mut r = t.range((s + 1)..).map_err(|_| INKONSISTENT)?;
            match r.next() {
                Some(e) => Some(e.map_err(|_| INKONSISTENT)?.0.value()),
                None => None,
            }
        }
        Richtung::Voriger => {
            let Some(s) = satz else {
                return Ok(t.last().map_err(|_| INKONSISTENT)?.map(|(k, _)| k.value()));
            };
            let mut r = t.range(..s).map_err(|_| INKONSISTENT)?;
            match r.next_back() {
                Some(e) => Some(e.map_err(|_| INKONSISTENT)?.0.value()),
                None => None,
            }
        }
    };
    Ok(id)
}

/// Bewegung in der Ordnung eines Index. Der Ausgangspunkt wird aus dem
/// aktuellen Satz frisch berechnet — deshalb übersteht der Cursor jede
/// Satzänderung (D5).
fn index_bewegen(
    txn: &WriteTransaction,
    tabelle: &str,
    def: &Indexdef,
    layout: &Layout,
    udts: &[UdtLayout],
    satz: Option<u64>,
    richtung: Richtung,
) -> Result<Option<u64>, RuntimeError> {
    let name = index_def(tabelle, &def.name);
    let idx: IndexTab = TableDefinition::new(&name);
    let t = txn.open_table(idx).map_err(|_| INKONSISTENT)?;
    let wert = |e: Option<(redb::AccessGuard<'_, &[u8]>, redb::AccessGuard<'_, u64>)>| {
        e.map(|(_, v)| v.value())
    };
    match richtung {
        Richtung::Erster => Ok(wert(t.first().map_err(|_| INKONSISTENT)?)),
        Richtung::Letzter => Ok(wert(t.last().map_err(|_| INKONSISTENT)?)),
        Richtung::Naechster | Richtung::Voriger => {
            let vorwaerts = richtung == Richtung::Naechster;
            let Some(s) = satz else {
                // Unpositioniert: von vorn bzw. von hinten anfangen.
                return Ok(wert(if vorwaerts {
                    t.first().map_err(|_| INKONSISTENT)?
                } else {
                    t.last().map_err(|_| INKONSISTENT)?
                }));
            };
            // Schlüssel des aktuellen Satzes rekonstruieren.
            let bytes = satz_bytes_lesen(txn, tabelle, s)?.ok_or(KEIN_SATZ)?;
            let felder = felder_lesen(&bytes, layout, udts)?;
            let k = indexschluessel(def, &felder, layout, s);
            if vorwaerts {
                let bereich: (std::ops::Bound<&[u8]>, std::ops::Bound<&[u8]>) = (
                    std::ops::Bound::Excluded(k.as_slice()),
                    std::ops::Bound::Unbounded,
                );
                let mut r = t.range::<&[u8]>(bereich).map_err(|_| INKONSISTENT)?;
                let treffer = match r.next() {
                    Some(e) => Some(e.map_err(|_| INKONSISTENT)?.1.value()),
                    None => None,
                };
                Ok(treffer)
            } else {
                let mut r = t.range(..k.as_slice()).map_err(|_| INKONSISTENT)?;
                let treffer = match r.next_back() {
                    Some(e) => Some(e.map_err(|_| INKONSISTENT)?.1.value()),
                    None => None,
                };
                Ok(treffer)
            }
        }
    }
}

#[cfg(test)]
mod tests;