`services/search/run_local.py`

99 lines
"""Runnable anonymous-search HTTP endpoint on the Python stdlib (no third-party deps, $0).

This is the dependency-free transport over SearchService — it lets the whole anti-enumeration
+ masking path run and be smoke-tested without FastAPI/uvicorn/OpenSearch. Production uses
the FastAPI reference in search_api.py; this proves the core end-to-end over real HTTP.

    python -m services.search.run_local            # seeds synthetic data, serves :8799
    curl 'http://127.0.0.1:8799/v1/search?q=Oneil'
    curl 'http://127.0.0.1:8799/healthz'

Env: UPP_PORT (default 8799), UPP_RATE_MAX (default 30), UPP_RATE_WINDOW (default 60).
Client-key = the direct peer IP (trusted for a local direct-connect server). Behind a real
proxy, derive the key from the trusted proxy hop, never a raw X-Forwarded-For header.
"""
from __future__ import annotations

import json
import os
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlparse

from services.common.rate_limit import SlidingWindowRateLimiter
from services.common.sqlite_repo import FileObjectStore, SqliteRepository
from services.ingestion.ingest import FeedDefinition, ingest_authorized_feed
from services.search.service import EmptyQuery, RateLimited, SearchService

_SAMPLE = Path(__file__).resolve().parents[2] / "data" / "sample" / "sample_state_feed.csv"


def build_service() -> SearchService:
    import tempfile
    tmp = Path(tempfile.mkdtemp(prefix="upp-serve-"))
    store = FileObjectStore(tmp / "obj")
    (tmp / "obj").mkdir(parents=True, exist_ok=True)
    store.write_bytes("incoming/s.csv", _SAMPLE.read_bytes())
    repo = SqliteRepository(str(tmp / "serve.db"))
    ingest_authorized_feed(FeedDefinition("SAMPLE", "incoming/s.csv"), store, repo)
    limiter = SlidingWindowRateLimiter(
        max_requests=int(os.environ.get("UPP_RATE_MAX", "30")),
        window_seconds=float(os.environ.get("UPP_RATE_WINDOW", "60")),
    )
    return SearchService(repository=repo, limiter=limiter)


def make_handler(service: SearchService):
    class Handler(BaseHTTPRequestHandler):
        def _json(self, code: int, body: dict, headers: dict | None = None) -> None:
            payload = json.dumps(body).encode()
            self.send_response(code)
            self.send_header("Content-Type", "application/json")
            for k, v in (headers or {}).items():
                self.send_header(k, str(v))
            self.end_headers()
            self.wfile.write(payload)

        def do_GET(self) -> None:  # noqa: N802
            parsed = urlparse(self.path)
            if parsed.path == "/healthz":
                self._json(200, {"status": "ok"})
                return
            if parsed.path != "/v1/search":
                self._json(404, {"error": "not found"})
                return
            qs = parse_qs(parsed.query)
            query = (qs.get("q", [""])[0])
            client_key = f"ip:{self.client_address[0]}"
            try:
                results = service.search(query, client_key=client_key)
                self._json(200, {"results": results, "count": len(results)})
            except EmptyQuery as e:
                self._json(400, {"error": str(e)})
            except RateLimited as e:
                self._json(429, {"error": "rate limited"},
                           headers={"Retry-After": int(e.retry_after) + 1})

        def log_message(self, *args) -> None:  # quiet
            pass

    return Handler


def main() -> int:
    port = int(os.environ.get("UPP_PORT", "8799"))
    service = build_service()
    # Single-threaded: the SQLite connection is bound to this thread. Requests serialize,
    # which is fine for a local demo. Production (Postgres + connection pool) is thread-safe.
    httpd = HTTPServer(("127.0.0.1", port), make_handler(service))
    print(f"serving synthetic unclaimed-property search on http://127.0.0.1:{port}")
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass
    return 0


if __name__ == "__main__":
    raise SystemExit(main())