"""The collector control plane (Stream AW4.2; the brief `.logs/planning/streams/AW4-live-page.md`, D1).

Schema only, server defaults, no CHECK constraints, no backfill: a row a running collector holds
is never touched, and a container that started before the upgrade keeps collecting.

`collection_runs` gains what the live page reads and the loop writes: `heartbeat_at` and
`requests_made` (the hook, at every request boundary and every control poll), `pid` and
`started_by` (the child, at run creation), `mode` and `limit_n` (`limit` is a reserved word),
`expected_total`, the seven counters the loop keeps after resolution, and `stopped_by`. The
`status` column is a plain varchar, so the new value `stopped` needs no schema change; it is
named here so the migration says what the code now writes.

`sources` gains the cooperative control the page sets and the loop reads between requests
(`control`, who set it and when), the pace a person set (`delay_set_by`, `delay_set_at`), the
host's robots crawl delay as the loop last read it (a published fact, never a control), and the
mode the next Start uses.

Why: on 19 Sep four `running` rows outlived the processes an OOM killed, and the approval gate
held the review for the 24 hours the timer had left. A heartbeat per request tells a live run from
a dead one in an hour, and a control field the loop reads makes start, pause, resume and stop page
actions instead of `docker exec`.

Revision ID: aw4a1b2c3d4e
Revises: aw2c1d2e3f4a
Create Date: 2026-09-19
"""

import sqlalchemy as sa

from alembic import op

revision = "aw4a1b2c3d4e"
down_revision = "aw2c1d2e3f4a"
branch_labels = None
depends_on = None

COUNTERS = (
    "existing_checked", "existing_changed", "existing_missing",
    "new_found", "new_brands", "new_lines", "new_variants",
)

RUN_COLUMNS = (
    sa.Column("heartbeat_at", sa.DateTime(timezone=True)),
    sa.Column("requests_made", sa.Integer(), nullable=False, server_default=sa.text("0")),
    sa.Column("pid", sa.Integer()),
    sa.Column("mode", sa.String(12), nullable=False, server_default="discover"),
    sa.Column("limit_n", sa.Integer()),
    sa.Column("expected_total", sa.Integer()),
    *(sa.Column(name, sa.Integer(), nullable=False, server_default=sa.text("0")) for name in COUNTERS),
    sa.Column("started_by", sa.String(80)),
    sa.Column("stopped_by", sa.String(80)),
)

SOURCE_COLUMNS = (
    sa.Column("control", sa.String(8), nullable=False, server_default="run"),
    sa.Column("control_set_by", sa.String(80)),
    sa.Column("control_set_at", sa.DateTime(timezone=True)),
    sa.Column("delay_set_by", sa.String(80)),
    sa.Column("delay_set_at", sa.DateTime(timezone=True)),
    sa.Column("robots_crawl_delay", sa.Numeric(6, 2)),
    sa.Column("robots_read_at", sa.DateTime(timezone=True)),
    sa.Column("mode", sa.String(12), nullable=False, server_default="discover"),
)


def upgrade() -> None:
    for column in RUN_COLUMNS:
        op.add_column("collection_runs", column)
    for column in SOURCE_COLUMNS:
        op.add_column("sources", column)
    # The live read asks for the running rows on every poll; the status column had no index.
    op.create_index(
        "ix_collection_runs_running", "collection_runs", ["source_id"],
        postgresql_where=sa.text("status = 'running'"),
    )


def downgrade() -> None:
    op.drop_index("ix_collection_runs_running", table_name="collection_runs")
    for column in reversed(SOURCE_COLUMNS):
        op.drop_column("sources", column.name)
    for column in reversed(RUN_COLUMNS):
        op.drop_column("collection_runs", column.name)