# -*- coding: utf-8 -*-
"""t-260930083937: идемпотентная автодоставка артефактов в TG (hermetic, без LLM/TG).

Запуск: ./venv/bin/python tests/test_delivery_idempotency.py

Дубли в чат (m-516173239/241 — «Ты мне её уже три раза отправил»)
приходили из гонки уровней: push в тике и delivery-демон оба проходили
проверку delivery_status != 'sent', пока TG-отправка была в полёте
(23–34с между дублями DeliverySent в сторе). Плюс ad-hoc curl sendDocument
вообще без метки delivered — каждый тик слал файл заново.

Проверяет: ledger (первая запись только одна), атомарный claim
(живой/истёкший лок, release, межпроцессная атомарность), детерминированную
гонку deliver_result (push в полёте + параллельный демон → одна отправка),
SKIP с событием, сбой TG → release → досылка демоном, requeue со сбросом
ledger и флагов модели, pre-filter deliver_pending, deliver_artifact
(dedup/force/сбой+ретрай), CLI tg-notify -k / tg-deliver-file (--force).
"""
import contextlib
import io
import json
import os
import subprocess
import sys
import threading

# ── изоляция: своя БД в /tmp (prod не трогаем) ──
_DST = "/tmp/simbiot-test-delivery-idem.db"
for p in (_DST, _DST + "-wal", _DST + "-shm"):
    if os.path.exists(p):
        os.unlink(p)
os.environ["SIMBIOT_DB"] = _DST
sys.path.insert(0, "/root/simbiot-harness")

from harness import es, tg_inbox
from harness import commands
from harness.scheduler import deliver_pending

fails = []


def check(name, cond, extra=""):
    print(("PASS" if cond else "FAIL"), name, ("| " + str(extra)) if not cond else "")
    if not cond:
        fails.append(name)


def tg_sends():
    """Все события фактической отправки (DeliverySent/ArtifactDelivered) в сторе."""
    db = es._conn()
    try:
        rows = db.execute(
            "SELECT aggregate_id, event_type FROM events "
            "WHERE event_type IN ('DeliverySent','ArtifactDelivered')").fetchall()
        return [(r["aggregate_id"], r["event_type"]) for r in rows]
    finally:
        db.close()


def count_skips(aggregate_id):
    db = es._conn()
    try:
        return db.execute(
            "SELECT COUNT(*) AS n FROM events WHERE aggregate='task' "
            "AND aggregate_id=? AND event_type='DeliverySkipped'",
            (aggregate_id,)).fetchone()["n"]
    finally:
        db.close()


def run_cli(*argv):
    """CLI-команда in-process (stdout перехвачен) — моки tg_inbox действуют."""
    import cli as cli_mod
    old = sys.argv
    sys.argv = ["cli.py"] + list(argv)
    buf = io.StringIO()
    try:
        with contextlib.redirect_stdout(buf):
            cli_mod.main()
    finally:
        sys.argv = old
    return buf.getvalue()


# ═══════════════ 1. ledger: доставлено ровно один раз ═══════════════
es.init_db()
aid = "file:/tmp/pres.html"
check("ledger: первая запись", es.record_delivery(aid, "file") is True)
check("ledger: дубль записи отклонён", es.record_delivery(aid, "file") is False)
check("ledger: recorded видит запись", es.delivery_recorded(aid) is True)
check("ledger: отсутствующий не recorded",
      es.delivery_recorded("file:/tmp/none.html") is False)

# ═══════════════ 2. claim: атомарный захват ДО отправки ═══════════════
check("claim: первый захват", es.delivery_claim("lock-1", owner="A", ttl_sec=60) is True)
check("claim: живой лок блокирует", es.delivery_claim("lock-1", owner="B", ttl_sec=60) is False)
es.delivery_claim("lock-exp", owner="A", ttl_sec=-1)  # уже истёк
check("claim: истёкший лок перехватывается",
      es.delivery_claim("lock-exp", owner="B", ttl_sec=60) is True)
es.delivery_release("lock-1")
check("claim: release открывает повторный захват",
      es.delivery_claim("lock-1", owner="B", ttl_sec=60) is True)

# межпроцессная атомарность: два дочерних процесса тянут один claim
code = ("import os,sys;os.environ['SIMBIOT_DB']=%r;"
        "sys.path.insert(0,'/root/simbiot-harness');"
        "from harness import es;"
        "print('1' if es.delivery_claim('lock-xproc', owner='child', ttl_sec=60) else '0')"
        % _DST)
r1 = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True)
r2 = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True)
wins = (r1.stdout.strip() + r2.stdout.strip()).replace("\n", "")
check("claim: межпроцессно ровно один победитель", sorted(wins) == ["0", "1"], wins)

# ═══════════════ 3. deliver_artifact: dedup / force / сбой+ретрай ═══════════════
calls = []
ok1 = commands.deliver_artifact("art-1", "file", lambda: (calls.append(1) or True),
                                owner="test", caption="презентация")
ok2 = commands.deliver_artifact("art-1", "file", lambda: (calls.append(1) or True),
                                owner="test")
check("artifact: первая доставка True", ok1 is True)
check("artifact: повтор False", ok2 is False)
check("artifact: send вызван один раз", len(calls) == 1, f"calls={len(calls)}")
check("artifact: ledger записан", es.delivery_recorded("art-1") is True)
check("artifact: событие ArtifactDelivered есть",
      ("art-1", "ArtifactDelivered") in tg_sends())

okf = commands.deliver_artifact("art-1", "file", lambda: (calls.append(1) or True),
                                owner="test", force=True)
check("artifact: force пробивает живой лок", okf is True)
check("artifact: force — второй send", len(calls) == 2, f"calls={len(calls)}")
check("artifact: force не «молодит» ledger (IGNORE)",
      es.record_delivery("art-1", "file") is False)

flaky = {"n": 0}
def _flaky():
    flaky["n"] += 1
    if flaky["n"] == 1:
        raise RuntimeError("TG упал")
    return True
try:
    commands.deliver_artifact("art-flaky", "file", _flaky, owner="test")
    check("artifact: сбой прокинут", False)
except RuntimeError:
    check("artifact: сбой прокинут", True)
check("artifact: сбой не записан в ledger",
      es.delivery_recorded("art-flaky") is False)
check("artifact: ретрай после сбоя успешен",
      commands.deliver_artifact("art-flaky", "file", _flaky, owner="test") is True)

# ═══════════════ 4. deliver_result: гонка уровней = одна отправка ═══════════════
# инцидент: push в тике и delivery-демон параллельно → 2× DeliverySent.
# Детерминированная гонка: поток A (тик) берёт claim и «висит в отправке»
# (как TG до 35с), B (демон) идёт в это же время → claim отклоняет.
tid, _ = commands.submit_task("Задача с автодоставкой", "детали", 5)
commands.mark_awaiting_delivery(tid, reason="тест")
es.append_event("task", tid, "TaskClaimed", {"level": "hour", "session_id": "s-test"})
es.append_event("task", tid, "TaskCompleted", {"answer": "ответ задачи"})

_orig_notify = tg_inbox.notify_chat
in_flight = threading.Event()   # держим «отправку» потока A
started = threading.Event()     # A дошёл до notify

def _slow_notify(text):
    started.set()
    in_flight.wait(timeout=10)  # имитация полёта TG-отправки
    return True

res_a = {}
def _deliver_a():
    res_a["r"] = commands.deliver_result(tid, "ответ задачи", owner="tick:hour")

tg_inbox.notify_chat = _slow_notify
try:
    ta = threading.Thread(target=_deliver_a)
    ta.start()
    check("гонка: A дошёл до отправки (claim взят)", started.wait(timeout=5))
    # параллельный уровень/демон в то же время: claim жив → пропуск
    res_b = commands.deliver_result(tid, "ответ задачи", owner="daemon")
    in_flight.set()             # отпускаем A: отправка завершается
    ta.join(timeout=10)
finally:
    tg_inbox.notify_chat = _orig_notify

sent = [a for a, t in tg_sends() if a == tid and t == "DeliverySent"]
check("гонка: ровно один DeliverySent", len(sent) == 1, f"sent={len(sent)}")
check("гонка: B (демон) пропущен claim'ом", res_b is False, f"res_b={res_b}")
check("гонка: A доставил (True)", res_a.get("r") is True, f"res_a={res_a}")
check("гонка: ledger записан", es.delivery_recorded(tid) is True)

# повтор при записанном ledger (stale read model): SKIP с событием
before = count_skips(tid)
check("stale: повтор SKIP", commands.deliver_result(tid, "ответ") is False)
check("stale: SKIP залогирован DeliverySkipped",
      count_skips(tid) == before + 1, f"skips={count_skips(tid)}")
model = es.run_projections()
check("stale: проекция delivery_skips видна",
      model["tasks"][tid].get("delivery_skips") == 1,
      model["tasks"][tid].get("delivery_skips"))

# ═══════════════ 5. сбой TG: release → демон досылает ═══════════════
tid2, _ = commands.submit_task("Задача с падением TG", "детали", 5)
commands.mark_awaiting_delivery(tid2, reason="тест")
es.append_event("task", tid2, "TaskClaimed", {"level": "hour", "session_id": "s-test2"})
es.append_event("task", tid2, "TaskCompleted", {"answer": "ответ 2"})

def _dead_notify(text):
    raise RuntimeError("TG недоступен")
tg_inbox.notify_chat = _dead_notify
try:
    check("сбой TG: deliver_result False",
          commands.deliver_result(tid2, "ответ 2", owner="daemon") is False)
finally:
    tg_inbox.notify_chat = _orig_notify
check("сбой TG: ledger пуст", es.delivery_recorded(tid2) is False)
db = es._conn()
try:
    row = db.execute("SELECT COUNT(*) AS n FROM delivery_locks WHERE artifact_id=?",
                     (tid2,)).fetchone()
    check("сбой TG: лок освобождён (не ждём TTL)", row["n"] == 0, row["n"])
finally:
    db.close()
# TG ожил: демон досылает на следующем проходе
tg_inbox.notify_chat = lambda text: True
try:
    delivered = deliver_pending()
finally:
    tg_inbox.notify_chat = _orig_notify
check("сбой TG: демон дослал на след. проходе", tid2 in delivered, delivered)
check("сбой TG: DeliverySent появился",
      (tid2, "DeliverySent") in tg_sends())

# ═══════════════ 6. requeue: сброс ledger и флагов → новая доставка ═══════════════
commands.requeue_task(tid, "доработка по просьбе Alex")
check("requeue: ledger сброшен", es.delivery_recorded(tid) is False)
model = es.run_projections()
check("requeue: флаг модели сброшен в await (не 'sent' — досылка не порезана)",
      model["tasks"][tid].get("delivery_status") == "await",
      model["tasks"][tid].get("delivery_status"))
check("requeue: задача снова ready", model["tasks"][tid]["status"] == "ready")
# доработали: новый результат должен уйти (мок)
es.append_event("task", tid, "TaskClaimed", {"level": "hour", "session_id": "s-rq"})
es.append_event("task", tid, "TaskCompleted", {"answer": "ответ после доработки"})
tg_inbox.notify_chat = lambda text: True
try:
    redone = commands.deliver_result(tid, "ответ после доработки", owner="tick:hour")
finally:
    tg_inbox.notify_chat = _orig_notify
check("requeue: новый результат доставлен", redone is True)
check("requeue: после доработки ровно +1 DeliverySent (2 всего)",
      len([1 for a, t in tg_sends() if a == tid and t == "DeliverySent"]) == 2)

# ═══════════════ 7. deliver_pending: pre-filter без дублей ═══════════════
tg_inbox.notify_chat = lambda text: True
try:
    delivered2 = deliver_pending()
finally:
    tg_inbox.notify_chat = _orig_notify
check("pending: уже доставленное не досылается", delivered2 == [], f"got={delivered2}")
# 2 задачи: tid дважды (гонка + доработка после requeue) и tid2 раз → 3
check("pending: DeliverySent не размножены (3 всего: 2×tid + 1×tid2)",
      len([1 for a, t in tg_sends() if t == "DeliverySent"]) == 3)

# ═══════════════ 8. CLI: tg-notify -k / tg-deliver-file ═══════════════
tg_inbox.notify_chat = lambda text: True
tg_inbox.send_document_to_chat = lambda path, caption="": True
try:
    r1 = run_cli("tg-notify", "тестовое уведомление", "-k", "cli-key-1")
    r2 = run_cli("tg-notify", "тестовое уведомление", "-k", "cli-key-1")
finally:
    tg_inbox.notify_chat = _orig_notify
check("cli: tg-notify -k первая — OK", "OK" in r1, r1)
check("cli: tg-notify -k повтор — SKIP", "SKIP" in r2, r2)
check("cli: ledger по ключу записан",
      es.delivery_recorded("msg:cli-key-1") is True)

open("/tmp/simbiot-test-dl.txt", "w").write("файл-артефакт")
_orig_send_doc = tg_inbox.send_document_to_chat
tg_inbox.notify_chat = lambda text: True
tg_inbox.send_document_to_chat = lambda path, caption="": True
try:
    r3 = run_cli("tg-deliver-file", "/tmp/simbiot-test-dl.txt", "-c", "подпись")
    r4 = run_cli("tg-deliver-file", "/tmp/simbiot-test-dl.txt", "-c", "подпись")
    r5 = run_cli("tg-deliver-file", "/tmp/simbiot-test-dl.txt", "-c", "повтор",
                 "--force")
finally:
    tg_inbox.notify_chat = _orig_notify
    tg_inbox.send_document_to_chat = _orig_send_doc
expected_aid = f"file:{os.path.abspath('/tmp/simbiot-test-dl.txt')}"
check("cli: tg-deliver-file первая — OK", "OK" in r3, r3)
check("cli: tg-deliver-file повтор — SKIP", "SKIP" in r4, r4)
check("cli: --force пробивает SKIP", "OK" in r5, r5)
check("cli: ledger по файлу записан",
      es.delivery_recorded(expected_aid) is True)

# ═══════════════ итог ═══════════════
print()
if fails:
    print(f"FAILED: {len(fails)} проверок провалено: {fails}")
    sys.exit(1)
print("OK: идемпотентная доставка — все проверки прошли")