"""Processes and threads that make up one run of the WAL starvation lab.

Imported by run.py. Python 3 standard library only. Every process opens its
own sqlite3 connection with isolation_level=None, so the transactions below
are exactly the ones written here and nothing is opened behind our back.

Timing: every process waits for the same wall-clock start time t0, then
measures durations with time.perf_counter(). Timestamps written to the CSVs
are seconds since t0.
"""

import os
import sqlite3
import time
from array import array

INSERT = "INSERT INTO t(ts, b) VALUES (?, ?)"
BLOB_BYTES = 1000


def wait_until(t_wall):
    delay = t_wall - time.time()
    if delay > 0:
        time.sleep(delay)


def writer(path, t0, duration, rate, journal_size_limit, autocheckpoint, out):
    """One writer, open loop: commit i is due at t0 + i/rate.

    If a commit stalls, the following commits are late but are still sent,
    so a stall shows up as latency instead of as a quietly lower rate.
    Each commit is one autocommit INSERT of a 1000-byte blob.
    Python's default connect timeout (5 s) is the busy handler.
    """
    con = sqlite3.connect(path, isolation_level=None, timeout=5.0)
    con.execute("PRAGMA synchronous=NORMAL")
    if journal_size_limit is not None:
        con.execute(f"PRAGMA journal_size_limit={int(journal_size_limit)}")
    if autocheckpoint is not None:
        con.execute(f"PRAGMA wal_autocheckpoint={int(autocheckpoint)}")
    settings = {
        "journal_mode": con.execute("PRAGMA journal_mode").fetchone()[0],
        "synchronous": con.execute("PRAGMA synchronous").fetchone()[0],
        "wal_autocheckpoint": con.execute("PRAGMA wal_autocheckpoint").fetchone()[0],
        "journal_size_limit": con.execute("PRAGMA journal_size_limit").fetchone()[0],
        "page_size": con.execute("PRAGMA page_size").fetchone()[0],
    }
    blob = bytes((i * 131 + 7) % 256 for i in range(BLOB_BYTES))  # fixed bytes, SQLite does not compress
    start = array("d")    # commit start, seconds since t0
    service = array("q")  # time inside execute(), microseconds
    response = array("q")  # completion minus scheduled time, microseconds
    errors = 0
    wait_until(t0)
    p0 = time.perf_counter()
    i = 0
    while True:
        due = i / rate
        if due >= duration:
            break
        now = time.perf_counter() - p0
        if now < due:
            time.sleep(due - now)
        s = time.perf_counter()
        try:
            con.execute(INSERT, (time.time(), blob))
        except sqlite3.OperationalError:
            errors += 1
        e = time.perf_counter()
        start.append(s - p0)
        service.append(int((e - s) * 1e6))
        response.append(int((e - p0 - due) * 1e6))
        i += 1
    con.close()
    out.put(("writer", {
        "settings": settings,
        "errors": errors,
        "start": start.tobytes(),
        "service": service.tobytes(),
        "response": response.tobytes(),
    }))


def reader(path, t0, idx, hold, idle, offset, stop, open_count, out):
    """One reader. Read transaction k opens at t0 + offset + k*(hold+idle).

    Inside it: BEGIN, SELECT max(id) (this takes the snapshot and pins a WAL
    read mark, or slot 0 if the WAL is fully checkpointed), stay open for `hold` seconds, look up the 100 newest rows of
    that snapshot and time it, COMMIT. Then no transaction until the next
    scheduled start.
    """
    con = sqlite3.connect(path, isolation_level=None, timeout=5.0)
    period = hold + idle
    rows = []
    k = 0
    while not stop.is_set():
        wait_until(t0 + offset + k * period)
        k += 1
        if stop.is_set():
            break
        with open_count.get_lock():
            open_count.value += 1
        t_open = time.time() - t0
        con.execute("BEGIN")
        max_id = con.execute("SELECT max(id) FROM t").fetchone()[0] or 0
        stop.wait(max(0.0, t0 + t_open + hold - time.time()))
        s = time.perf_counter()
        got = con.execute(
            "SELECT id, length(b) FROM t WHERE id > ? ORDER BY id LIMIT 100",
            (max_id - 100,),
        ).fetchall()
        lookup_us = int((time.perf_counter() - s) * 1e6)
        con.execute("COMMIT")
        t_close = time.time() - t0
        with open_count.get_lock():
            open_count.value -= 1
        rows.append((idx, round(t_open, 4), round(t_close, 4), max_id, len(got), lookup_us))
    con.close()
    out.put(("reader", rows))


def maintenance(path, t0, duration, period, mode, stop, rows):
    """Maintenance thread: PRAGMA wal_checkpoint(<mode>) every `period` s,
    at t0 + period, t0 + 2*period, ... while that time is before the end.

    Runs in the parent process on its own connection with a 5 s busy
    timeout. Appends (t_start, busy, log, checkpointed, duration_ms).
    """
    con = sqlite3.connect(path, isolation_level=None, timeout=5.0)
    k = 1
    while k * period < duration and not stop.wait(max(0.0, t0 + k * period - time.time())):
        t_start = time.time() - t0
        s = time.perf_counter()
        busy, log, ckpt = con.execute(f"PRAGMA wal_checkpoint({mode})").fetchone()
        ms = (time.perf_counter() - s) * 1e3
        rows.append((round(t_start, 3), mode, busy, log, ckpt, round(ms, 2)))
        k += 1
    con.close()


def wal_salts(wal_path):
    """Salt-1 and salt-2 from the WAL header (bytes 16..23, big-endian).

    walRestartHdr() adds 1 to salt-1 every time the WAL is reset to frame 1,
    and the header is rewritten when frame 1 is written, so watching salt-1
    counts resets. Returns None when the file is shorter than a header
    (right after a TRUNCATE checkpoint).
    """
    try:
        with open(wal_path, "rb") as f:
            hdr = f.read(32)
    except FileNotFoundError:
        return None
    if len(hdr) < 32:
        return None
    return int.from_bytes(hdr[16:20], "big"), int.from_bytes(hdr[20:24], "big")


def file_size(p):
    try:
        return os.path.getsize(p)
    except FileNotFoundError:
        return 0