fix: validate search alias checkpoints

backend/search/minjust_opensearch.py

@@ -15,7 +15,7 @@ import urllib.request
 from pathlib import Path
 from typing import Iterator
 
-APP_VERSION = "0.5.3"
+APP_VERSION = "0.6.0"
 LANGUAGES = {"ru", "ky"}
 DEFAULT_MAPPING = Path(__file__).with_name("minjust-fragments-index.json")
 
@@ -282,7 +282,15 @@ def checkpoint_state(
         "last_document_code",
         "complete",
     }
-    if not isinstance(state, dict) or set(state) != expected:
+    legacy_expected = expected - {"alias"}
+    if not isinstance(state, dict):
+        raise ValueError(f"Invalid checkpoint: {checkpoint}")
+    if set(state) == legacy_expected:
+        if state["schema_version"] != 1 or alias is not None:
+            raise ValueError(f"Legacy checkpoint does not support --alias: {checkpoint}")
+        state["schema_version"] = 2
+        state["alias"] = None
+    elif set(state) != expected:
         raise ValueError(f"Invalid checkpoint: {checkpoint}")
     if (
         state["schema_version"] != 2
@@ -319,7 +327,7 @@ def load_bulk(
     alias: str | None = None,
 ) -> tuple[int, int]:
     base = url.rstrip("/")
-    if alias is not None and alias == index:
+    if alias is not None and (not alias or alias == index):
         raise ValueError("--alias must differ from --index")
     index_url = f"{base}/{urllib.parse.quote(index, safe='')}"
     if resume:
@@ -419,7 +427,9 @@ def main() -> int:
         raise SystemExit("--resume requires --url")
     if arguments.checkpoint and not arguments.url:
         raise SystemExit("--checkpoint requires --url")
-    if arguments.alias and not arguments.url:
+    if arguments.alias == "":
+        raise SystemExit("--alias must not be empty")
+    if arguments.alias is not None and not arguments.url:
         raise SystemExit("--alias requires --url")
     if arguments.url:
         checkpoint = arguments.checkpoint or Path("data/opensearch") / f"{arguments.index}.checkpoint.json"