"""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()