m2-market/indexer/src/m2_market_indexer/watch.py

"""Incremental market:catalog sync: poll registry main tip, upsert deltas.

contracts/catalog-index.md: "watch — on registry merge (webhook or poll ≤5 min),
upsert changed listings, remove delisted." State (last-seen commit + per-listing
content hash) is cached in a state file so a no-op poll costs one API call.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import sys
import time
from pathlib import Path

import httpx
from jsonschema import Draft202012Validator

from .client import MemoryClient, RegistryClient, build_embedded_text
from .config import IndexerConfig, listing_schema_path, load_config


def _load_validator() -> Draft202012Validator:
    with listing_schema_path().open() as f:
        schema = json.load(f)
    return Draft202012Validator(schema)


def load_state(path: Path) -> dict:
    if not path.is_file():
        return {"last_commit": None, "listings": {}}
    with path.open() as f:
        return json.load(f)


def save_state(path: Path, state: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(".json.tmp")
    with tmp.open("w") as f:
        json.dump(state, f, indent=2)
    tmp.replace(path)


def _content_hash(listing: dict) -> str:
    return hashlib.sha256(
        json.dumps(listing, sort_keys=True).encode("utf-8")
    ).hexdigest()


def poll_once(
    registry: RegistryClient,
    memory: MemoryClient,
    validator: Draft202012Validator,
    state: dict,
) -> tuple[dict, dict]:
    """One poll cycle. Returns (new_state, summary). No-ops if the tip is unchanged."""
    tip = registry.get_main_commit_sha()
    if tip == state.get("last_commit"):
        return state, {"changed": False, "upserted": [], "deleted": []}

    prev_listings = state.get("listings", {})
    new_listings: dict = {}
    upserted: list[str] = []
    deleted: list[str] = []
    seen_ids: set[str] = set()

    for listing_id in registry.list_listing_ids():
        seen_ids.add(listing_id)
        try:
            listing = registry.get_listing(listing_id)
        except (httpx.HTTPStatusError, ValueError) as exc:
            print(f"WARN: skipping {listing_id}: cannot read listing.json ({exc})", file=sys.stderr)
            # Leave any prior state entry as-is; retried on the next poll.
            if listing_id in prev_listings:
                new_listings[listing_id] = prev_listings[listing_id]
            continue

        errors = list(validator.iter_errors(listing))
        if errors:
            print(
                f"WARN: skipping {listing_id}: schema violations: "
                + "; ".join(e.message for e in errors),
                file=sys.stderr,
            )
            if listing_id in prev_listings:
                new_listings[listing_id] = prev_listings[listing_id]
            continue

        status = listing.get("status")
        content_hash = _content_hash(listing)
        prev = prev_listings.get(listing_id)

        if status == "published":
            changed = (
                prev is None
                or prev.get("hash") != content_hash
                or prev.get("status") != "published"
            )
            if changed:
                solution = registry.get_solution(listing_id)
                text = build_embedded_text(listing, solution)
                memory.upsert_document(listing_id, text, listing)
                upserted.append(listing_id)
        elif prev is not None and prev.get("status") == "published":
            memory.delete_document(listing_id)
            deleted.append(listing_id)

        new_listings[listing_id] = {"status": status, "hash": content_hash}

    # Listing directories that disappeared entirely from the registry.
    for listing_id, meta in prev_listings.items():
        if listing_id not in seen_ids and meta.get("status") == "published":
            memory.delete_document(listing_id)
            deleted.append(listing_id)

    new_state = {"last_commit": tip, "listings": new_listings}
    return new_state, {"changed": True, "upserted": upserted, "deleted": deleted}


def _run_loop(config: IndexerConfig, interval: int) -> None:
    validator = _load_validator()
    state = load_state(config.state_path)

    with RegistryClient(
        config.forgejo_url,
        config.forgejo_token,
        config.registry_repo,
        config.registry_ref,
    ) as registry, MemoryClient(
        config.memory_api_url,
        config.memory_api_key,
        config.memory_api_partition,
        partition_path=config.memory_api_partition_path,
        document_path=config.memory_api_document_path,
    ) as memory:
        while True:
            state, summary = poll_once(registry, memory, validator, state)
            save_state(config.state_path, state)
            if summary["changed"]:
                print(
                    f"watch: upserted={summary['upserted']} deleted={summary['deleted']}"
                )
            else:
                print("watch: no change")
            time.sleep(interval)


def main() -> None:
    parser = argparse.ArgumentParser(prog="m2-market-indexer-watch")
    parser.add_argument("--interval", type=int, default=None, help="poll interval in seconds")
    args = parser.parse_args()

    config = load_config()
    interval = args.interval if args.interval is not None else config.poll_interval_seconds
    _run_loop(config, interval)


if __name__ == "__main__":
    main()