Merge branch 'feature/ogm-endpoints' into develop

Author: ewlarson

AGENTS.md

@@ -99,8 +99,11 @@ Run from the **project root**. Key targets:
 | `make test-fast`       | Backend tests, parallel, no coverage. |
 | `make test-fresh-db`   | Recreate test DB from main DB. |
 | `make lint-test`       | `lint-check` then `test`. |
-| `make reindex`         | Reindex all resources into Elasticsearch (via API container). |
+| `make reindex`         | Atomic local reindex using versioned index + alias swap, then clear local search cache. |
+| `make reindex-benchmark` | Run atomic local reindex with benchmark timing output enabled. |
+| `make kamal-reindex`   | Atomic reindex on Kamal with versioned index + alias swap; auto-runs `kamal-clear-cache`. |
 | `make verify-h3-index` | Verify H3 pyramid fields in Elasticsearch. |
+| `make kamal-clear-cache` | Clear remote API cache on Kamal (`KAMAL_CACHE_TYPE`, default `search`). |
 | `make clear_cache`     | Flush Redis cache (needs `REDIS_PASSWORD` in `.env`). |
 | `make frontend-reset`  | Clear Vite cache and restart frontend-dev. |
 | `make db-export`       | Export ParadeDB to `tmp/btaa_geospatial_api_export.sql.gz`. |
@@ -132,17 +135,21 @@ See `docs/make_tasks.md` for overrides (e.g. `COVERAGE_THRESHOLD`, `PARALLEL_WOR
 
 ### Indexing (Elasticsearch)
 
-- **Reindex all resources** (same behavior as `/admin/reindex`):  
+- **Reindex all resources locally** (safe atomic cutover):  
   `make reindex`  
   or  
-  `docker compose exec api bash -lc "cd /app/backend && python scripts/reindex_admin.py"`
+  `docker compose exec api bash -lc "cd /app/backend && python scripts/reindex_atomic.py"`
+
+- **Atomic reindex on Kamal** (safe cutover):  
+  `make kamal-reindex`  
+  Builds a versioned index, atomically swaps alias `ELASTICSEARCH_INDEX`, keeps one previous versioned index by default, then clears remote API cache.
 
 - **Verify H3 pyramid fields** after reindex:  
   `make verify-h3-index`  
   or  
   `docker compose exec api bash -lc "cd /app/backend && python scripts/verify_h3_index.py"`
 
-Backend scripts for one-off or debug indexing: `backend/scripts/reindex.py`, `simple_bulk_index.py`, `run_index.py`. See `docs/backend/scripts.md` and `docs/backend/ogm_harvesting.md`.
+Backend scripts for one-off or debug indexing: `backend/scripts/reindex.py`, `backend/scripts/reindex_atomic.py`, `simple_bulk_index.py`, `run_index.py`. See `docs/backend/scripts.md` and `docs/backend/ogm_harvesting.md`.
 
 ---
 

Makefile

@@ -1,4 +1,4 @@
-.PHONY: lint lint-check format test lint-test test-coverage-compare db-export db-import db-sync gbl-admin-db-download gbl-admin-db-unzip gbl-admin-db-restore gbl-admin-db-sync gbl-admin-db-add-latest-btaa-fields gbl-admin-db-import-resources populate-distributions populate-data-dictionaries gbl-admin-db-import-all reindex es-unblock populate-relationships verify-h3-index kamal-reindex kamal-verify-h3-index clear_cache frontend-reset
+.PHONY: lint lint-check format test lint-test test-coverage-compare db-export db-import db-sync gbl-admin-db-download gbl-admin-db-unzip gbl-admin-db-restore gbl-admin-db-sync gbl-admin-db-add-latest-btaa-fields gbl-admin-db-import-resources populate-distributions populate-data-dictionaries gbl-admin-db-import-all reindex reindex-benchmark local-clear-search-cache es-unblock populate-relationships verify-h3-index kamal-reindex kamal-verify-h3-index kamal-clear-cache clear_cache frontend-reset ogm-refresh ogm-refresh-all ogm-refresh-repo ogm-status ogm-status-watch ogm-failures
 
 # Load environment variables from .env file if it exists
 -include .env
@@ -47,6 +47,33 @@ GBL_ADMIN_IMPORT_CONFLICT ?= update
 GBL_ADMIN_DISTRIBUTIONS_BATCH_SIZE ?= 2000
 KAMAL_APP_ROLE ?= web
 KAMAL_PYTHON ?= /opt/venv/bin/python
+# Local atomic reindex tuning.
+REINDEX_CHUNK_SIZE ?= 2000
+REINDEX_BULK_SIZE ?= 2000
+REINDEX_BULK_MAX_RETRIES ?= 2
+REINDEX_FAST_SETTINGS ?= true
+REINDEX_FORCE_REPLICAS_ZERO ?= true
+REINDEX_ALLOW_PARTIAL ?= false
+REINDEX_PRUNE_OLD ?= true
+REINDEX_RETAIN_PREVIOUS ?= 1
+REINDEX_REMOVE_LEGACY_INDEX ?= true
+REINDEX_BENCHMARK ?= false
+# Kamal atomic reindex defaults (versioned index + alias swap + prune).
+KAMAL_REINDEX_CHUNK_SIZE ?= 1000
+KAMAL_REINDEX_BULK_SIZE ?= 2000
+KAMAL_REINDEX_BULK_MAX_RETRIES ?= 2
+KAMAL_REINDEX_FAST_SETTINGS ?= true
+KAMAL_REINDEX_FORCE_REPLICAS_ZERO ?= true
+KAMAL_REINDEX_ALLOW_PARTIAL ?= false
+KAMAL_REINDEX_PRUNE_OLD ?= true
+KAMAL_REINDEX_RETAIN_PREVIOUS ?= 1
+KAMAL_REINDEX_REMOVE_LEGACY_INDEX ?= true
+# Optional override for remote API base URL used by kamal-clear-cache.
+# If unset, the target falls back to APPLICATION_URL from Kamal env.
+KAMAL_API_URL ?=
+KAMAL_CACHE_TYPE ?= search
+OGM_API_URL ?= http://localhost:8000
+OGM_STATUS_POLL_SECONDS ?= 5
 
 # Run both linting and formatting checks (without modifying files)
 lint:
@@ -451,14 +478,110 @@ es-unblock:
 		-d '{"index.blocks.read_only_allow_delete": null}' || true
 	@echo "Done. Run 'make reindex' if you need to reindex."
 
-# Reindex all resources into Elasticsearch
+# Reindex all resources into Elasticsearch using versioned index + alias swap.
 reindex:
-	@echo "Reindexing all resources into Elasticsearch (same logic as /admin/reindex)..."
+	@echo "Atomic reindex (versioned index + alias swap) in local Docker..."
 	@docker compose exec -T api bash -lc '\
 		set -e; \
 		: $${ELASTICSEARCH_INDEX:=btaa_geospatial_api}; \
-		echo "Index: $$ELASTICSEARCH_INDEX"; \
-		cd /app/backend && python3 scripts/reindex_admin.py'
+		echo "Alias: $$ELASTICSEARCH_INDEX"; \
+		cd /app/backend && \
+		REINDEX_ATOMIC_CHUNK_SIZE=$(REINDEX_CHUNK_SIZE) \
+		REINDEX_ATOMIC_BULK_SIZE=$(REINDEX_BULK_SIZE) \
+		REINDEX_ATOMIC_BULK_MAX_RETRIES=$(REINDEX_BULK_MAX_RETRIES) \
+		REINDEX_ATOMIC_FAST_SETTINGS=$(REINDEX_FAST_SETTINGS) \
+		REINDEX_ATOMIC_FORCE_REPLICAS_ZERO=$(REINDEX_FORCE_REPLICAS_ZERO) \
+		REINDEX_ATOMIC_ALLOW_PARTIAL=$(REINDEX_ALLOW_PARTIAL) \
+		REINDEX_ATOMIC_PRUNE_OLD=$(REINDEX_PRUNE_OLD) \
+		REINDEX_ATOMIC_RETAIN_PREVIOUS=$(REINDEX_RETAIN_PREVIOUS) \
+		REINDEX_ATOMIC_REMOVE_LEGACY_INDEX=$(REINDEX_REMOVE_LEGACY_INDEX) \
+		REINDEX_ATOMIC_BENCHMARK=$(REINDEX_BENCHMARK) \
+		python3 scripts/reindex_atomic.py'
+	@echo "Reindex complete; clearing local search cache..."
+	@$(MAKE) local-clear-search-cache
+
+# Reindex with benchmark timing output enabled.
+reindex-benchmark:
+	@$(MAKE) reindex REINDEX_BENCHMARK=true
+
+# Clear local API search cache via admin endpoint.
+local-clear-search-cache:
+	@docker compose exec -T api bash -lc 'ADMIN_USER=$${ADMIN_USERNAME:-admin}; ADMIN_PASS=$${ADMIN_PASSWORD:-changeme}; curl -fsS -u "$$ADMIN_USER:$$ADMIN_PASS" -X POST "http://localhost:8000/api/v1/admin/cache/clear?cache_type=search"'
+
+# Refresh OpenGeoMetadata (OGM) harvest for all enabled weekly repos.
+# Uses ADMIN_USERNAME/ADMIN_PASSWORD from env or .env (defaults to admin/changeme).
+ogm-refresh: ogm-refresh-all
+
+ogm-refresh-all:
+	@echo "Triggering OGM harvest for all enabled weekly repos via $(OGM_API_URL)..."
+	@docker compose exec -T api bash -lc '\
+		ADMIN_USER=$${ADMIN_USERNAME:-admin}; \
+		ADMIN_PASS=$${ADMIN_PASSWORD:-changeme}; \
+		curl -fsS -u "$$ADMIN_USER:$$ADMIN_PASS" -X POST \
+			"$(OGM_API_URL)/api/v1/admin/ogm/harvest" \
+			-H "Content-Type: application/json" \
+			-d '\''{"ogm_all":true,"ogm_trigger":"weekly"}'\'''
+	@echo
+	@echo "OGM harvest request submitted."
+
+# Refresh a single OpenGeoMetadata (OGM) repo.
+# Usage: make ogm-refresh-repo OGM_REPO_NAME=edu.stanford.purl
+ogm-refresh-repo:
+	@if [ -z "$(OGM_REPO_NAME)" ]; then \
+		echo "ERROR: OGM_REPO_NAME is required."; \
+		echo "Usage: make ogm-refresh-repo OGM_REPO_NAME=edu.stanford.purl"; \
+		exit 1; \
+	fi
+	@echo "Triggering OGM harvest for repo $(OGM_REPO_NAME) via $(OGM_API_URL)..."
+	@docker compose exec -T api bash -lc '\
+		ADMIN_USER=$${ADMIN_USERNAME:-admin}; \
+		ADMIN_PASS=$${ADMIN_PASSWORD:-changeme}; \
+		curl -fsS -u "$$ADMIN_USER:$$ADMIN_PASS" -X POST \
+			"$(OGM_API_URL)/api/v1/admin/ogm/harvest" \
+			-H "Content-Type: application/json" \
+			-d "{\"ogm_repo_name\":\"$(OGM_REPO_NAME)\",\"ogm_trigger\":\"manual\"}"'
+	@echo
+	@echo "OGM harvest request submitted for $(OGM_REPO_NAME)."
+
+# Show OGM harvest status snapshot.
+# Usage:
+#   make ogm-status                      # list recent runs
+#   make ogm-status OGM_RUN_ID=<run_id> # show one run detail
+ogm-status:
+	@echo "Fetching OGM harvest status from $(OGM_API_URL)..."
+	@docker compose exec -T api bash -lc '\
+		ADMIN_USER=$${ADMIN_USERNAME:-admin}; \
+		ADMIN_PASS=$${ADMIN_PASSWORD:-changeme}; \
+		if [ -n "$(OGM_RUN_ID)" ]; then \
+			curl -fsS -u "$$ADMIN_USER:$$ADMIN_PASS" \
+				"$(OGM_API_URL)/api/v1/admin/ogm/harvest/runs/$(OGM_RUN_ID)"; \
+		else \
+			curl -fsS -u "$$ADMIN_USER:$$ADMIN_PASS" \
+				"$(OGM_API_URL)/api/v1/admin/ogm/harvest/runs"; \
+		fi'
+	@echo
+
+# Continuously poll OGM harvest status.
+# Usage:
+#   make ogm-status-watch
+#   make ogm-status-watch OGM_RUN_ID=<run_id> OGM_STATUS_POLL_SECONDS=3
+ogm-status-watch:
+	@echo "Watching OGM harvest status (every $(OGM_STATUS_POLL_SECONDS)s). Press Ctrl+C to stop."
+	@while true; do \
+		$(MAKE) --no-print-directory ogm-status OGM_RUN_ID="$(OGM_RUN_ID)"; \
+		sleep $(OGM_STATUS_POLL_SECONDS); \
+	done
+
+# Show only failed OGM harvest runs with error details.
+# Usage: make ogm-failures
+ogm-failures:
+	@echo "Fetching failed OGM harvest runs from $(OGM_API_URL)..."
+	@docker compose exec -T api bash -lc '\
+		ADMIN_USER=$${ADMIN_USERNAME:-admin}; \
+		ADMIN_PASS=$${ADMIN_PASSWORD:-changeme}; \
+		curl -fsS -u "$$ADMIN_USER:$$ADMIN_PASS" \
+			"$(OGM_API_URL)/api/v1/admin/ogm/harvest/runs" \
+		| python -c "import json,sys; data=json.load(sys.stdin); failed=[r for r in data.get(\"runs\",[]) if r.get(\"ogm_status\")==\"failed\"]; print(\"No failed OGM runs found in current history window.\") if not failed else [print(\"ogm_id={0} repo={1} trigger={2} started_at={3}\\nerror={4}\\n\".format(r.get(\"ogm_id\"), r.get(\"ogm_repo_name\"), r.get(\"ogm_trigger\"), r.get(\"ogm_started_at\"), r.get(\"ogm_error\") or \"(none)\")) for r in failed]"'
 
 # Populate resource_relationships from resources table (dct_isPartOf_sm, pcdm_memberOf_sm, etc.).
 # Run after ingest or when relationship columns change. Search "Has part" filter uses DB + ES;
@@ -494,13 +617,25 @@ verify-h3-index:
 
 # Reindex all resources into Elasticsearch on Kamal (single role run by default).
 kamal-reindex:
-	@echo "Reindexing all resources on Kamal (role: $(KAMAL_APP_ROLE))..."
+	@echo "Atomic reindex on Kamal (role: $(KAMAL_APP_ROLE))..."
 	@if [ -z "$$KAMAL_SSH_USER" ] || [ -z "$$KAMAL_HOST" ]; then \
 		echo "ERROR: KAMAL_SSH_USER and KAMAL_HOST environment variables must be set."; \
 		echo "Please source .kamal/secrets first."; \
 		exit 1; \
 	fi
-	@kamal app exec --roles $(KAMAL_APP_ROLE) "bash -lc 'cd /app/backend && $(KAMAL_PYTHON) scripts/reindex_admin.py'"
+	@kamal app exec --roles $(KAMAL_APP_ROLE) "bash -lc 'cd /app/backend && \
+		REINDEX_ATOMIC_CHUNK_SIZE=$(KAMAL_REINDEX_CHUNK_SIZE) \
+		REINDEX_ATOMIC_BULK_SIZE=$(KAMAL_REINDEX_BULK_SIZE) \
+		REINDEX_ATOMIC_BULK_MAX_RETRIES=$(KAMAL_REINDEX_BULK_MAX_RETRIES) \
+		REINDEX_ATOMIC_FAST_SETTINGS=$(KAMAL_REINDEX_FAST_SETTINGS) \
+		REINDEX_ATOMIC_FORCE_REPLICAS_ZERO=$(KAMAL_REINDEX_FORCE_REPLICAS_ZERO) \
+		REINDEX_ATOMIC_ALLOW_PARTIAL=$(KAMAL_REINDEX_ALLOW_PARTIAL) \
+		REINDEX_ATOMIC_PRUNE_OLD=$(KAMAL_REINDEX_PRUNE_OLD) \
+		REINDEX_ATOMIC_RETAIN_PREVIOUS=$(KAMAL_REINDEX_RETAIN_PREVIOUS) \
+		REINDEX_ATOMIC_REMOVE_LEGACY_INDEX=$(KAMAL_REINDEX_REMOVE_LEGACY_INDEX) \
+		$(KAMAL_PYTHON) scripts/reindex_atomic.py'"
+	@echo "Reindex complete; clearing API cache (cache_type: $(KAMAL_CACHE_TYPE))..."
+	@$(MAKE) kamal-clear-cache
 
 # Verify H3 pyramid fields on Kamal (single role run by default).
 kamal-verify-h3-index:
@@ -512,6 +647,22 @@ kamal-verify-h3-index:
 	fi
 	@kamal app exec --roles $(KAMAL_APP_ROLE) "bash -lc 'cd /app/backend && $(KAMAL_PYTHON) scripts/verify_h3_index.py'"
 
+# Clear API cache on Kamal via admin endpoint (defaults to search cache).
+# Usage:
+#   make kamal-clear-cache
+#   make kamal-clear-cache KAMAL_CACHE_TYPE=all
+#   make kamal-clear-cache KAMAL_CACHE_TYPE=suggest
+kamal-clear-cache:
+	@echo "Clearing remote cache on Kamal (role: $(KAMAL_APP_ROLE), cache_type: $(KAMAL_CACHE_TYPE))..."
+	@if [ -z "$$KAMAL_SSH_USER" ] || [ -z "$$KAMAL_HOST" ]; then \
+		echo "ERROR: KAMAL_SSH_USER and KAMAL_HOST environment variables must be set."; \
+		echo "Please source .kamal/secrets first."; \
+		exit 1; \
+	fi
+	@kamal app exec --roles $(KAMAL_APP_ROLE) "bash -lc 'ADMIN_USER=\$${ADMIN_USERNAME:-admin}; ADMIN_PASS=\$${ADMIN_PASSWORD:-changeme}; API_BASE=\"$(KAMAL_API_URL)\"; if [ -z \"\$$API_BASE\" ]; then API_BASE=\"\$$APPLICATION_URL\"; fi; if [ -z \"\$$API_BASE\" ]; then echo \"ERROR: KAMAL_API_URL or APPLICATION_URL must be set.\"; exit 1; fi; API_BASE=\"\$${API_BASE%/}\"; curl -fsS -u \"\$$ADMIN_USER:\$$ADMIN_PASS\" -X POST \"\$$API_BASE/api/v1/admin/cache/clear?cache_type=$(KAMAL_CACHE_TYPE)\"'"
+	@echo
+	@echo "Remote cache clear request submitted."
+
 # (Old index_missing_resources target removed; resilient reindex handles verification/repair)
 
 # Frontend (Docker dev): clear Vite cache and restart dev server.

backend/app/api/v1/endpoint_modules/ogm.py

@@ -0,0 +1,76 @@
+from typing import Optional
+
+from fastapi import APIRouter, Query
+
+from app.api.v1.utils import create_response
+from app.services.ogm_harvest.repository import OGMHarvestRepository
+
+router = APIRouter()
+ogm_repo = OGMHarvestRepository()
+
+
+@router.get("/ogm/repos")
+async def list_public_ogm_repos():
+    """
+    Public OGM repo summaries with latest crawl status and harvest counts.
+    """
+    repos = await ogm_repo.list_public_repo_summaries()
+    return create_response({"repos": repos})
+
+
+@router.get("/ogm/harvest/failures")
+async def list_public_ogm_harvest_failures(
+    repo_name: Optional[str] = Query(None, description="Filter by a single ogm_repo_name"),
+    include_with_errors: bool = Query(
+        True,
+        description=(
+            "Include runs that completed with non-zero import errors (not only ogm_status=failed)"
+        ),
+    ),
+    limit: int = Query(50, ge=1, le=500),
+    offset: int = Query(0, ge=0),
+):
+    """
+    Public OGM harvest failure runs, optionally filtered to a single repo.
+    """
+    if not include_with_errors:
+        failures = await ogm_repo.list_harvest_runs(
+            ogm_repo_name=repo_name,
+            ogm_status="failed",
+            limit=limit,
+            offset=offset,
+        )
+    else:
+        # Pull a bounded recent window, then filter runs that are failed OR have import errors.
+        window = await ogm_repo.list_harvest_runs(
+            ogm_repo_name=repo_name,
+            ogm_status=None,
+            limit=min(500, limit + offset + 200),
+            offset=0,
+        )
+        filtered = []
+        for run in window:
+            stats = run.get("ogm_stats_json") or {}
+            import_errors = int(stats.get("errors") or 0)
+            status = (run.get("ogm_status") or "").lower()
+            if status == "failed" or import_errors > 0:
+                filtered.append(run)
+        failures = filtered[offset : offset + limit]
+
+    for run in failures:
+        stats = run.get("ogm_stats_json") or {}
+        run["import_error_count"] = int(stats.get("errors") or 0)
+        run["imported_count"] = int(stats.get("imported") or 0)
+        run["error_samples"] = list(stats.get("error_samples") or [])[:10]
+        run["error_signatures"] = list(stats.get("error_signatures") or [])[:10]
+        run["failure_reason"] = run.get("ogm_error") or (
+            "One or more records failed during import" if run["import_error_count"] > 0 else None
+        )
+
+    return create_response(
+        {
+            "failures": failures,
+            "repo_name": repo_name,
+            "include_with_errors": include_with_errors,
+        }
+    )

backend/app/api/v1/endpoint_modules/root.py

@@ -51,6 +51,8 @@ async def api_root(request: Request = None):
                 "/thumbnails/placeholder",
                 "/thumbnails/{image_hash}",
                 "/map/h3",
+                "/ogm/repos",
+                "/ogm/harvest/failures",
                 "/mcp",
             ],
         },

backend/app/api/v1/endpoints.py

@@ -6,6 +6,7 @@
 from .endpoint_modules.gazetteer import router as gazetteer_router
 from .endpoint_modules.map import router as map_router
 from .endpoint_modules.mcp import router as mcp_router
+from .endpoint_modules.ogm import router as ogm_router
 from .endpoint_modules.ogm_webhook import router as ogm_webhook_router
 from .endpoint_modules.resources import router as resources_router
 
@@ -28,6 +29,7 @@
 router.include_router(static_maps_router, tags=["static-maps"])
 router.include_router(map_router, tags=["map"])
 router.include_router(mcp_router, tags=["mcp"])
+router.include_router(ogm_router, tags=["ogm"])
 # Hide admin, gazetteer, and shapefiles endpoints from Swagger documentation
 router.include_router(admin_router, prefix="/admin", tags=["admin"], include_in_schema=False)
 router.include_router(ogm_webhook_router, prefix="/admin", tags=["admin"], include_in_schema=False)