"""Ingestion guards: co-registrant filings (shared accessions), poller termination (form filter, Redis seen-set,
bounded retries), failure samples, conditional statement rewrite, coverage dictionary, schema migrations."""
from __future__ import annotations

import types
from datetime import date

import httpx
import pytest
from sqlalchemy import delete, select

PARENT, SUB = 65984, 66901            # Entergy Corp + Entergy Arkansas: one 10-Q/8-K, two registrants
SHARED = "0000065984-26-000283"
OTHER = "0000065984-26-000290"


def _atom(entries: list[tuple[str, str, int, str]]) -> str:
    items = []
    for form, name, cik, accn in entries:
        nodash = accn.replace("-", "")
        items.append(f"""<entry>
<title>{form} - {name} ({cik:010d}) (Filer)</title>
<link rel="alternate" type="text/html" href="https://www.sec.gov/Archives/edgar/data/{cik}/{nodash}/{accn}-index.htm"/>
<updated>2026-09-04T17:27:57-04:00</updated>
<category scheme="https://www.sec.gov/" label="form type" term="{form}"/>
<id>urn:tag:sec.gov,2008:accession-number={accn}</id>
</entry>""")
    return ('<?xml version="1.0" encoding="ISO-8859-1" ?>\n<feed xmlns="http://www.w3.org/2005/Atom">\n<title>Latest Filings</title>\n'
            + "\n".join(items) + "\n</feed>")


class FakeClient:
    """Minimal EdgarClient stand-in for poll_new_filings: canned Atom feed, optional daily index."""

    def __init__(self, atom: str, index: str | None = None):
        self.atom, self.index = atom, index
        self.stats = types.SimpleNamespace(requests=0)
        self.index_calls: list[date] = []

    def atom_current(self, form: str = "") -> str:
        self.stats.requests += 1
        return self.atom

    def daily_index(self, d: date) -> str | None:
        self.stats.requests += 1
        self.index_calls.append(d)
        return self.index


@pytest.fixture
def cofilers(fundamentals_data):
    """Two temporary tracked companies sharing accessions; everything they touch is removed afterwards."""
    from core.db import session
    from fundamentals import ingest
    from fundamentals.models import EdgarCompany, EdgarFiling, FundIngestState
    from stream.broker import get_redis
    r = get_redis()
    r.delete(ingest.SEEN_KEY, ingest.FAILED_KEY)
    with session() as s:
        before = s.get(FundIngestState, "incremental")
        snapshot = {k: getattr(before, k) for k in ("failures", "failure_samples")} if before else None
        s.add(EdgarCompany(cik=PARENT, ticker="ETR", tickers=["ETR"], name="Entergy Corp", status="active", ticker_history=[]))
        s.add(EdgarCompany(cik=SUB, ticker="EAL", tickers=["EAL"], name="Entergy Arkansas", status="active", ticker_history=[]))
    try:
        yield {"parent": PARENT, "sub": SUB, "redis": r}
    finally:
        from fundamentals.models import FundCoverage, fund_statements
        with session() as s:
            s.execute(delete(EdgarFiling).where(EdgarFiling.cik.in_([PARENT, SUB])))
            s.execute(delete(EdgarCompany).where(EdgarCompany.cik.in_([PARENT, SUB])))
            s.execute(delete(FundCoverage).where(FundCoverage.cik.in_([PARENT, SUB])))
            s.execute(delete(fund_statements).where(fund_statements.c.cik.in_([PARENT, SUB])))
            st = s.get(FundIngestState, "incremental")
            if st is not None and snapshot is not None:
                st.failures, st.failure_samples = snapshot["failures"], snapshot["failure_samples"]
        r.delete(ingest.SEEN_KEY, ingest.FAILED_KEY)


def _stub(monkeypatch, outcome):
    """Replace ingest_company with a recorder; `outcome(cik)` → error string or None."""
    from fundamentals import ingest
    calls: list[int] = []

    def fake(client, cik, **kw):
        calls.append(int(cik))
        res = ingest.IngestResult(cik=int(cik), ticker=str(cik))
        res.error = outcome(int(cik))
        res.changed = res.error is None
        return res
    monkeypatch.setattr(ingest, "ingest_company", fake)
    return calls


# ------------------------------------------------------------------------------------- co-registrants
def test_shared_accession_is_one_row_per_registrant(cofilers):
    import pandas as pd

    from fundamentals import ingest
    from fundamentals import normalize as N
    from fundamentals import service as S
    shared = N.Filing(accn=SHARED, form="10-Q", filed=date(2026, 7, 31), report_date=date(2026, 6, 30), primary_doc="etr-20260630.htm")
    only_parent = N.Filing(accn=OTHER, form="8-K", filed=date(2026, 8, 1), report_date=date(2026, 8, 1))
    untracked = N.Filing(accn="0000065984-26-000299", form="S-8", filed=date(2026, 8, 2))
    assert ingest._upsert_filings(PARENT, {SHARED: shared, OTHER: only_parent, "x": untracked}, pd.DataFrame()) == 2
    assert ingest._upsert_filings(SUB, {SHARED: shared}, pd.DataFrame()) == 1          # no IntegrityError any more
    assert ingest._upsert_filings(SUB, {SHARED: shared}, pd.DataFrame()) == 0          # idempotent
    pairs = {(PARENT, SHARED), (SUB, SHARED), (PARENT, OTHER), (SUB, OTHER)}
    assert ingest._known_pairs(pairs) == {(PARENT, SHARED), (SUB, SHARED), (PARENT, OTHER)}
    assert ingest._known_accns(SUB) == {SHARED}
    # both companies list the shared filing with their own EDGAR links
    for tk, cik in (("ETR", PARENT), ("EAL", SUB)):
        df, meta = S.filings(tk, form=None, date_from=None, date_to=None)
        row = df[df["accn"] == SHARED].iloc[0]
        assert meta["cik"] == cik and row["primary_doc_url"] == f"https://www.sec.gov/Archives/edgar/data/{cik}/000006598426000283/etr-20260630.htm"
    assert len(S.filings("ETR", form="10-Q,8-K", date_from=None, date_to=None)[0]) == 2
    assert len(S.filings("EAL", form=None, date_from=None, date_to=None)[0]) == 1


def test_poll_ingests_each_coregistrant_once(cofilers, monkeypatch):
    from fundamentals import ingest
    calls = _stub(monkeypatch, lambda cik: None)
    atom = _atom([("8-K", "Entergy Corp", PARENT, SHARED), ("8-K", "Entergy Arkansas", SUB, SHARED),
                  ("S-8", "Entergy Corp", PARENT, "0000065984-26-000299"),        # untracked form: ignored
                  ("8-K", "Nobody Inc", 999999999, "0000999999-26-000001")])      # not a tracked CIK
    client = FakeClient(atom, index=None)
    s1 = ingest.poll_new_filings(client, forms=("8-K",))
    assert s1["seen"] == 2 and s1["new"] == 2 and s1["affected_ciks"] == [PARENT, SUB] and s1["index_days"] == []
    assert sorted(calls) == [PARENT, SUB] and s1["rows_changed"] == 2 and s1["failures"] == []
    assert all(d.weekday() < 5 for d in client.index_calls) and len(client.index_calls) == 2
    # the stub recorded nothing in edgar_filings: without the Redis guard the same pairs would be "new" for ever
    s2 = ingest.poll_new_filings(client, forms=("8-K",))
    assert s2["seen"] == 2 and s2["new"] == 0 and s2["affected_ciks"] == [] and len(calls) == 2
    assert cofilers["redis"].scard(ingest.SEEN_KEY) == 2 and cofilers["redis"].ttl(ingest.SEEN_KEY) > 6 * 86_400


def test_amendment_in_the_8k_feed_is_tracked(cofilers, monkeypatch):
    """The `type=8-K` Atom feed also returns 8-K/A entries — they must land in edgar_filings (Genasys loop)."""
    from fundamentals import ingest
    from fundamentals.edgar_client import TRACKED_FORMS
    assert {"8-K/A", "10-K/A", "10-Q/A", "10-KT", "10-QT", "20-F/A", "6-K/A"} <= set(TRACKED_FORMS)
    calls = _stub(monkeypatch, lambda cik: None)
    client = FakeClient(_atom([("8-K/A", "Entergy Corp", PARENT, OTHER)]))
    s1 = ingest.poll_new_filings(client, forms=("8-K",), include_daily_index=False)
    assert s1["new"] == 1 and calls == [PARENT]


def test_transient_failures_are_retried_then_abandoned(cofilers, monkeypatch):
    from core.db import session
    from fundamentals import ingest
    from fundamentals.models import FundIngestState
    calls = _stub(monkeypatch, lambda cik: "HTTPStatusError: 503 Service Unavailable" if cik == PARENT else None)
    client = FakeClient(_atom([("8-K", "Entergy Corp", PARENT, SHARED), ("8-K", "Entergy Arkansas", SUB, SHARED)]))
    with session() as s:
        st = s.get(FundIngestState, "incremental")
        failures0 = (st.failures or 0) if st else 0
    summaries = [ingest.poll_new_filings(client, forms=("8-K",), include_daily_index=False) for _ in range(ingest.MAX_ATTEMPTS + 1)]
    # the subsidiary succeeded once; the parent was retried MAX_ATTEMPTS times then marked seen
    assert calls.count(SUB) == 1 and calls.count(PARENT) == ingest.MAX_ATTEMPTS
    assert [s["new"] for s in summaries] == [2, 1, 1, 0]
    assert summaries[0]["failures"] == ["HTTPStatusError: 503 Service Unavailable"]
    with session() as s:
        st = s.get(FundIngestState, "incremental")
        assert st.failures == failures0 + ingest.MAX_ATTEMPTS
        # samples survive the clean 4th cycle and carry the CIK
        assert st.failure_samples and all(x.startswith(f"cik={PARENT} HTTPStatusError") for x in st.failure_samples[-ingest.MAX_ATTEMPTS:])
    assert cofilers["redis"].hlen(ingest.FAILED_KEY) == 0        # cleared once marked seen


def test_no_facts_is_permanent_not_a_failure(cofilers, monkeypatch):
    from core.db import session
    from fundamentals import ingest
    from fundamentals.models import FundIngestState
    calls = _stub(monkeypatch, lambda cik: ingest.NO_FACTS)
    client = FakeClient(_atom([("8-K", "Entergy Corp", PARENT, OTHER)]))
    with session() as s:
        st = s.get(FundIngestState, "incremental")
        failures0 = (st.failures or 0) if st else 0
    s1 = ingest.poll_new_filings(client, forms=("8-K",), include_daily_index=False)
    s2 = ingest.poll_new_filings(client, forms=("8-K",), include_daily_index=False)
    assert s1["new"] == 1 and s1["no_facts"] == [PARENT] and s1["failures"] == [] and s2["new"] == 0 and calls == [PARENT]
    with session() as s:
        assert (s.get(FundIngestState, "incremental").failures or 0) == failures0


def test_ingest_company_without_facts_still_records_filings(cofilers):
    """companyfacts 404 (funds, trusts…) must not leave the accessions unknown — the poller would loop on them."""
    from fundamentals import ingest
    sub = {"cik": PARENT, "name": "Entergy Corp", "fiscalYearEnd": "1231", "filings": {"recent": {
        "accessionNumber": [SHARED, OTHER], "form": ["10-Q", "8-K"], "filingDate": ["2026-07-31", "2026-08-01"],
        "reportDate": ["2026-06-30", "2026-08-01"], "primaryDocument": ["etr-20260630.htm", "etr-8k.htm"], "isXBRL": [1, 1]}}}
    res = ingest.ingest_company(None, PARENT, companyfacts={"cik": PARENT, "facts": {}}, submissions=sub)
    assert res.error is None or res.error == ingest.NO_FACTS
    no_facts_client = types.SimpleNamespace(companyfacts=lambda cik, refresh=False: None,
                                            submissions=lambda cik, refresh=False: {**sub, "cik": SUB})
    res = ingest.ingest_company(no_facts_client, SUB)
    assert res.error == ingest.NO_FACTS and not res.failed and res.filings == 2
    assert ingest._known_accns(SUB) == {SHARED, OTHER}


# ------------------------------------------------------------------------------------- daily index
def test_business_days_back():
    from fundamentals.ingest import business_days_back
    assert business_days_back(date(2026, 9, 6)) == [date(2026, 9, 4), date(2026, 9, 3)]      # Sunday → Fri, Thu
    assert business_days_back(date(2026, 9, 7)) == [date(2026, 9, 7), date(2026, 9, 4)]      # Monday → Mon, Fri
    assert business_days_back(date(2026, 9, 9), 3) == [date(2026, 9, 9), date(2026, 9, 8), date(2026, 9, 7)]


def test_daily_index_403_means_not_published(app, tmp_path):
    """www.sec.gov/Archives answers 403 for objects that do not exist (weekend indexes): no exception, no retry."""
    from fundamentals import edgar_client as ec
    hits: list[str] = []

    def handler(request: httpx.Request) -> httpx.Response:
        hits.append(str(request.url))
        assert "contact@" in request.headers["user-agent"] and "gzip" in request.headers["accept-encoding"]
        if request.url.path.endswith("master.20260906.idx"):
            return httpx.Response(403, text="<Error><Code>AccessDenied</Code></Error>")
        return httpx.Response(200, text="CIK|Company Name|Form Type|Date Filed|File Name\n"
                                        "65984|ENTERGY CORP|8-K|20260904|edgar/data/65984/0000065984-26-000290.txt\n")
    client = ec.EdgarClient(rate=1000, raw_dir=tmp_path, http=httpx.Client(transport=httpx.MockTransport(handler), headers={
        "User-Agent": "HF Market Data (test, contact@spboucher.ai)", "Accept-Encoding": "gzip, deflate"}))
    assert client.daily_index(date(2026, 9, 6)) is None
    assert len(hits) == 1 and hits[0].endswith("/Archives/edgar/daily-index/2026/QTR3/master.20260906.idx")
    text = client.daily_index(date(2026, 9, 4))
    from fundamentals.ingest import parse_master_index
    assert parse_master_index(text) == [{"cik": 65984, "name": "ENTERGY CORP", "form": "8-K", "filed": "2026-09-04",
                                         "accn": "0000065984-26-000290"}]
    with pytest.raises(httpx.HTTPStatusError):
        client.get_text("https://www.sec.gov/Archives/edgar/daily-index/2026/QTR3/master.20260906.idx")


# ------------------------------------------------------------------------------------- statements / coverage
def test_replace_statements_only_rewrites_on_change(fundamentals_data):
    from core.db import session
    from fundamentals import ingest
    from fundamentals import service as S
    from fundamentals.models import fund_statements
    cik = 320193
    original = S._rows(cik)
    ids0 = sorted(r["id"] for r in original)
    assert original and all(isinstance(r["coverage"], dict) for r in original) and "coverage_id" not in original[0]
    assert ingest._replace_statements(cik, original) is False                      # identical → untouched
    with session() as s:
        assert sorted(s.scalars(select(fund_statements.c.id).where(fund_statements.c.cik == cik))) == ids0
    changed = [dict(r) for r in original]
    target = next(r for r in changed if r["statement"] == "income" and r.get("revenue") is not None)
    target["revenue"] = target["revenue"] + 1.0
    assert ingest._replace_statements(cik, changed) is True
    again = S._rows(cik)
    assert sorted(r["id"] for r in again) != ids0
    hit = next(r for r in again if r["accn"] == target["accn"] and r["statement"] == "income" and r["fiscal_quarter"] == target["fiscal_quarter"])
    assert hit["revenue"] == target["revenue"] and hit["coverage"] == target["coverage"]
    with session() as s:   # written through the dictionary: inline JSON is NULL, the reference is set
        rows = s.execute(select(fund_statements.c.coverage, fund_statements.c.coverage_id).where(fund_statements.c.cik == cik)).all()
    assert rows and all(c is None and cid is not None for c, cid in rows)
    assert ingest._replace_statements(cik, original) is True                       # restore for the other tests
    assert S._rows(cik)[0]["coverage"] == original[0]["coverage"]


def test_coverage_dictionary_interns_and_hydrates(app):
    from core.db import session
    from fundamentals import coverage_store as CS
    from fundamentals.models import FundCoverageBlob
    a = {"interest_expense": {"reason": "no_mapped_tag"}, "ebitda": {"computed": "operating_income + d_and_a"}}
    b = {"ebitda": {"computed": "operating_income + d_and_a"}, "interest_expense": {"reason": "no_mapped_tag"}}   # same, other order
    ids = CS.intern([a, b, {}, a])
    assert ids[0] == ids[1] == ids[3] and ids[2] != ids[0] and CS.digest(a) == CS.digest(b)
    CS.reset_cache_for_tests()
    assert CS.intern([b]) == [ids[0]]                                     # found in the table, not re-inserted
    with session() as s:
        n = s.scalar(select(FundCoverageBlob.id).where(FundCoverageBlob.sha1 == CS.digest(a)))
    assert n == ids[0]
    rows = [{"accn": "x", "coverage": None, "coverage_id": ids[0]},          # dictionary reference
            {"accn": "y", "coverage": {"revenue": {"tag": "us-gaap:Revenues"}}, "coverage_id": None},   # legacy inline
            {"accn": "z", "coverage": None, "coverage_id": None}]
    CS.hydrate(rows)
    assert rows[0]["coverage"] == a and rows[1]["coverage"] == {"revenue": {"tag": "us-gaap:Revenues"}} and rows[2]["coverage"] == {}
    assert all("coverage_id" not in r for r in rows)
    assert CS.sha_of_ids({ids[0]}) == {ids[0]: CS.digest(a)}


def test_coverage_intern_survives_cache_eviction(app, monkeypatch):
    """Prod 2026-09-06: the 50 533rd distinct document raised KeyError because _remember() wiped the cache
    (CACHE_MAX) in the middle of intern(); ids must be resolved locally, not through the cache."""
    from fundamentals import coverage_store as CS
    CS.reset_cache_for_tests()
    monkeypatch.setattr(CS, "CACHE_MAX", 5)
    docs = [{"k%d" % i: {"reason": "r%d" % i}} for i in range(20)] + [{"k0": {"reason": "r0"}}]
    ids = CS.intern(docs)
    assert len(ids) == 21 and len(set(ids[:20])) == 20 and ids[20] == ids[0]
    assert CS.intern(docs) == ids                                          # second pass: same ids, no KeyError


def test_failure_samples_append_and_cap(app):
    from core.db import session
    from fundamentals import ingest
    from fundamentals.models import FundIngestState
    key = "test_samples"
    try:
        ingest._set_state(key, failures_add=2, failure_samples_add=[f"e{i}" for i in range(12)])
        ingest._set_state(key, last_run_at=None)                          # a clean cycle keeps the evidence
        with session() as s:
            st = s.get(FundIngestState, key)
            assert st.failures == 2 and st.failure_samples == [f"e{i}" for i in range(2, 12)]
        ingest._set_state(key, failure_samples=[])                        # explicit replace still possible
        with session() as s:
            assert s.get(FundIngestState, key).failure_samples == []
    finally:
        with session() as s:
            s.execute(delete(FundIngestState).where(FundIngestState.key == key))


# ------------------------------------------------------------------------------------- migrations
def test_migrate_filings_pk_from_legacy_table(tmp_path):
    from sqlalchemy import create_engine, text

    from fundamentals import migrations
    eng = create_engine(f"sqlite:///{tmp_path / 'legacy.db'}")
    with eng.begin() as con:
        con.execute(text("""CREATE TABLE edgar_filings (accn VARCHAR(24) NOT NULL, cik INTEGER NOT NULL, form VARCHAR(16) NOT NULL,
                            filed_date DATE NOT NULL, period_of_report DATE, primary_doc VARCHAR(512), is_amendment BOOLEAN NOT NULL,
                            is_xbrl BOOLEAN NOT NULL, parsed_at DATETIME, PRIMARY KEY (accn))"""))
        con.execute(text("CREATE INDEX ix_edgar_filings_cik ON edgar_filings (cik)"))
        con.execute(text("INSERT INTO edgar_filings VALUES ('0000065984-26-000283', 65984, '10-Q', '2026-07-31', '2026-06-30', NULL, 0, 1, NULL)"))
        con.execute(text("INSERT INTO edgar_filings VALUES ('0000320193-24-000069', 320193, '10-Q', '2024-05-03', '2024-03-30', NULL, 0, 1, NULL)"))
        assert migrations.filings_pk_is_composite(con) is False
        assert migrations.migrate_filings_pk(con) == 2
        assert migrations.filings_pk_is_composite(con) is True
        # the subsidiary's copy of the shared accession now fits
        con.execute(text("INSERT INTO edgar_filings (cik, accn, form, filed_date, is_amendment, is_xbrl) VALUES (66901, '0000065984-26-000283', '10-Q', '2026-07-31', 0, 1)"))
        assert con.execute(text("SELECT count(*) FROM edgar_filings WHERE accn='0000065984-26-000283'")).scalar() == 2
        names = {r[0] for r in con.execute(text("SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='edgar_filings'"))}
        assert {"ix_edgar_filings_accn", "ix_edgar_filings_cik_filed", "ix_edgar_filings_form", "ix_edgar_filings_filed_date"} <= names
        assert migrations.migrate_filings_pk(con) == 3                     # idempotent rerun keeps every row


def test_startup_schema_is_idempotent(fundamentals_data):
    from core.db import engine
    from fundamentals import migrations
    assert migrations.ensure_filings_schema() == "ok"
    assert migrations.ensure_statements_schema() is False
    assert migrations.ensure_auto_vacuum() is False                        # already switched when the test DB was young
    with engine.connect() as con:
        from sqlalchemy import text
        assert int(con.execute(text("PRAGMA auto_vacuum")).scalar()) == 2
    migrations.incremental_vacuum()