#!/usr/bin/env catnip
# Checkpoint transactionnel local avec SQLite et freeze/thaw.
#
# Le pipeline résume un échantillon OpenAlex local. Le premier run calcule le
# rapport, projette son struct vers un record sérialisable avec freeze et le
# valide après un commit SQLite. Les runs suivants font thaw puis reconstruisent
# le type nominal. Une empreinte SHA-256 de la source invalide automatiquement
# un checkpoint périmé.
#
# La transaction BEGIN IMMEDIATE + INSERT ... ON CONFLICT garantit qu'un lecteur
# ne voit jamais un payload partiel. Aucun serveur externe n'est nécessaire.
#
# Pour forcer un miss reproductible :
#   CATNIP_SQLITE_RESET=1 catnip codex/persistence/sqlite_checkpoint_pipeline.cat
#
# DEPS: orjson

sqlite3 = import('sqlite3')
orjson = import('orjson')
hashlib = import('hashlib')
os = import('os')
import('pathlib', 'Path')

script_dir = Path(META.file).parent
source_path = script_dir / 'data/openalex-works-sample.jsonl'
output_dir = script_dir / 'output'
output_dir.mkdir(exist_ok=True)
db_path = output_dir / 'sqlite_checkpoint.db'

CHECKPOINT_KEY = 'openalex-local-summary-v1'

struct Checkpoint {
    source_hash: str; total: int; cited_sum: int; by_year; by_type;

    avg_citations(self): float => {
        if self.total > 0 { round(self.cited_sum / self.total, 2) } else { 0.0 }
    }

    display(self) => {
        print(f"  works={self.total}  citations={self.cited_sum}  moyenne={self.avg_citations()}")
        print(f"  années={self.by_year}")
        print(f"  types={self.by_type}")
    }

    record(self) => {
        dict(
            source_hash=self.source_hash,
            total=self.total,
            cited_sum=self.cited_sum,
            by_year=self.by_year,
            by_type=self.by_type,
        )
    }
}

checkpoint_from_record = (record): Checkpoint => {
    Checkpoint(record['source_hash'], record['total'], record['cited_sum'], record['by_year'], record['by_type'])
}

source_bytes = source_path.read_bytes()
source_hash = hashlib.sha256(source_bytes).hexdigest()

nonempty_lines = (raw) => {
    lines = list()
    for line in raw.split(b'\n') {
        stripped = line.strip()
        if stripped { lines.append(stripped) }
    }
    lines
}

summarize = (raw, digest: str): Checkpoint => {
    total = 0
    cited_sum = 0
    by_year = dict()
    by_type = dict()

    for line in nonempty_lines(raw) {
        work = orjson.loads(line)
        # .get() couvre la clé absente, ?? la valeur null : une seule forme pour les deux.
        year = work.get('publication_year') ?? 'unknown'
        kind = work.get('type') ?? 'unknown'
        cited = work.get('cited_by_count') ?? 0

        total = total + 1
        cited_sum = cited_sum + cited
        by_year[year] = by_year.get(year, 0) + 1
        by_type[kind] = by_type.get(kind, 0) + 1
    }

    Checkpoint(digest, total, cited_sum, by_year, by_type)
}

connection = sqlite3.connect(str(db_path))
connection.execute('PRAGMA journal_mode=WAL')
connection.execute('PRAGMA synchronous=FULL')
connection.execute(
    'CREATE TABLE IF NOT EXISTS checkpoints (' +
        'key TEXT PRIMARY KEY, source_hash TEXT NOT NULL, payload_hash TEXT NOT NULL, ' +
        'payload BLOB NOT NULL, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP)'
)
connection.commit()

if os.getenv('CATNIP_SQLITE_RESET', '0') == '1' {
    connection.execute('DELETE FROM checkpoints WHERE key = ?', tuple(CHECKPOINT_KEY))
    connection.commit()
    print("⇒ Checkpoint SQLite supprimé à la demande")
}

row = connection.execute(
    'SELECT source_hash, payload_hash, payload FROM checkpoints WHERE key = ?',
    tuple(CHECKPOINT_KEY),
).fetchone()

hit = row != None and row[0] == source_hash

if hit {
    checkpoint = checkpoint_from_record(thaw(row[2]))
    payload_hash = row[1]
    print("⇒ Checkpoint hit : thaw du record + reconstruction du struct, calcul évité")
} else {
    reason = if row == None { "absent" } else { "empreinte source modifiée" }
    print(f"⇒ Checkpoint miss : {reason}")
    checkpoint = summarize(source_bytes, source_hash)
    # freeze sérialise les valeurs structurelles. Le type nominal reste local au
    # programme et est reconstruit explicitement après thaw.
    payload = freeze(checkpoint.record())
    payload_hash = hashlib.sha256(payload).hexdigest()

    # Écriture atomique : le verrou est acquis avant l'upsert et libéré au
    # commit. Un crash avant commit laisse l'ancien checkpoint intact.
    connection.execute('BEGIN IMMEDIATE')
    connection.execute(
        'INSERT INTO checkpoints(key, source_hash, payload_hash, payload) VALUES (?, ?, ?, ?) ' +
            'ON CONFLICT(key) DO UPDATE SET source_hash=excluded.source_hash, ' +
            'payload_hash=excluded.payload_hash, payload=excluded.payload, updated_at=CURRENT_TIMESTAMP',
        tuple(CHECKPOINT_KEY, source_hash, payload_hash, sqlite3.Binary(payload)),
    )
    connection.commit()
    print(f"  transaction commit : {len(payload)} octets gelés")
}

print()
print("⇒ Rapport restauré")
checkpoint.display()
connection.close()

# Vérification après fermeture/réouverture : on relit exactement ce qu'un autre
# processus verrait, puis on contrôle l'empreinte du BLOB avant thaw.
verification = sqlite3.connect(str(db_path))
stored = verification.execute(
    'SELECT source_hash, payload_hash, payload FROM checkpoints WHERE key = ?',
    tuple(CHECKPOINT_KEY),
).fetchone()
verification.close()

stored_hash = hashlib.sha256(stored[2]).hexdigest()
restored = checkpoint_from_record(thaw(stored[2]))
expected_total = len(nonempty_lines(source_bytes))
year_total = sum(restored.by_year.values())
type_total = sum(restored.by_type.values())

hash_ok = stored[0] == source_hash and stored[1] == stored_hash and stored[1] == payload_hash
shape_ok = restored.total == expected_total and year_total == restored.total and type_total == restored.total
print()
print("⇒ Vérification après réouverture")
print(f"  empreintes source + payload : {hash_ok}")
print(f"  totaux lignes + années + types : {shape_ok}")

# Le roundtrip ne se vérifie que sur le run qui calcule : au run hit, checkpoint
# et restored sortent du même BLOB et la comparaison serait tautologique. Un
# checkpoint restauré ne s'auto-valide pas, il se recoupe avec la source.
if hit {
    oracle_ok = hash_ok and shape_ok
    print("  roundtrip du struct : non applicable (aucun calcul de référence sur ce run)")
} else {
    roundtrip_ok = restored.source_hash == checkpoint.source_hash and restored.cited_sum == checkpoint.cited_sum and
        restored.by_year == checkpoint.by_year and
        restored.by_type == checkpoint.by_type
    oracle_ok = hash_ok and shape_ok and roundtrip_ok
    print(f"  roundtrip du struct : {roundtrip_ok}")
}
print(f"⇒ Oracle global SQLite : {oracle_ok}")
print(f"⇒ Base locale → {db_path}")