🔨 Add LMDB maintenance utility (#5)

* Add LMDB rotation script

* Update scripts/rotate_lmdb.py

Co-authored-by: Copilot <[email protected]>

---------

Co-authored-by: Copilot <[email protected]>

Author: Nativu5

scripts/USAGE.md

@@ -0,0 +1,44 @@
+# Scripts Usage
+
+This directory contains maintenance utilities for the project.
+
+## Table of Contents
+
+- [dump_lmdb.py](#dump_lmdbpy)
+- [rotate_lmdb.py](#rotate_lmdbpy)
+
+## dump_lmdb.py
+
+Dump records from an LMDB database as a JSON array. If no keys are provided, the script outputs every record. When keys are supplied, only the specified records are returned.
+
+### Usage
+
+Dump all entries:
+
+```bash
+python scripts/dump_lmdb.py /path/to/lmdb
+```
+
+Dump specific keys:
+
+```bash
+python scripts/dump_lmdb.py /path/to/lmdb key1 key2
+```
+
+## rotate_lmdb.py
+
+Delete LMDB records older than a given duration or remove all records.
+
+### Usage
+
+Delete entries older than 14 days:
+
+```bash
+python scripts/rotate_lmdb.py /path/to/lmdb 14d
+```
+
+Delete all entries:
+
+```bash
+python scripts/rotate_lmdb.py /path/to/lmdb all
+```

scripts/dump_lmdb.py

@@ -0,0 +1,58 @@
+import argparse
+from pathlib import Path
+from typing import Any, Iterable, List
+
+import lmdb
+import orjson
+
+
+def _decode_value(value: bytes) -> Any:
+    """Decode a value from LMDB to Python data."""
+    try:
+        return orjson.loads(value)
+    except orjson.JSONDecodeError:
+        return value.decode("utf-8", errors="replace")
+
+
+def _dump_all(txn: lmdb.Transaction) -> List[dict[str, Any]]:
+    """Return all records from the database."""
+    result: List[dict[str, Any]] = []
+    for key, value in txn.cursor():
+        result.append({"key": key.decode("utf-8"), "value": _decode_value(value)})
+    return result
+
+
+def _dump_selected(txn: lmdb.Transaction, keys: Iterable[str]) -> List[dict[str, Any]]:
+    """Return records for the provided keys."""
+    result: List[dict[str, Any]] = []
+    for key in keys:
+        raw = txn.get(key.encode("utf-8"))
+        if raw is not None:
+            result.append({"key": key, "value": _decode_value(raw)})
+    return result
+
+
+def dump_lmdb(path: Path, keys: Iterable[str] | None = None) -> None:
+    """Print selected or all key-value pairs from the LMDB database."""
+    env = lmdb.open(str(path), readonly=True, lock=False)
+    with env.begin() as txn:
+        if keys:
+            records = _dump_selected(txn, keys)
+        else:
+            records = _dump_all(txn)
+    env.close()
+
+    print(orjson.dumps(records, option=orjson.OPT_INDENT_2).decode())
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Dump LMDB records as JSON")
+    parser.add_argument("path", type=Path, help="Path to LMDB directory")
+    parser.add_argument("keys", nargs="*", help="Keys to retrieve")
+    args = parser.parse_args()
+
+    dump_lmdb(args.path, args.keys)
+
+
+if __name__ == "__main__":
+    main()

scripts/rotate_lmdb.py

@@ -0,0 +1,70 @@
+import argparse
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+import lmdb
+import orjson
+
+
+def _parse_duration(value: str) -> timedelta:
+    """Parse duration in the format '14d' or '24h'."""
+    if value.endswith("d"):
+        return timedelta(days=int(value[:-1]))
+    if value.endswith("h"):
+        return timedelta(hours=int(value[:-1]))
+    raise ValueError("Invalid duration format. Use Nd or Nh")
+
+
+def _should_delete(record: dict[str, Any], threshold: datetime) -> bool:
+    """Check if the record is older than the threshold."""
+    timestamp = record.get("updated_at") or record.get("created_at")
+    if not timestamp:
+        return False
+    try:
+        ts = datetime.fromisoformat(timestamp)
+    except ValueError:
+        return False
+    return ts < threshold
+
+
+def rotate_lmdb(path: Path, keep: str) -> None:
+    """Remove records older than the specified duration."""
+    env = lmdb.open(str(path), writemap=True, readahead=False, meminit=False)
+    if keep == "all":
+        with env.begin(write=True) as txn:
+            cursor = txn.cursor()
+            for key, _ in cursor:
+                txn.delete(key)
+        env.close()
+        return
+
+    delta = _parse_duration(keep)
+    threshold = datetime.now() - delta
+
+    with env.begin(write=True) as txn:
+        cursor = txn.cursor()
+        for key, value in cursor:
+            try:
+                record = orjson.loads(value)
+            except orjson.JSONDecodeError:
+                continue
+            if _should_delete(record, threshold):
+                txn.delete(key)
+    env.close()
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Remove outdated LMDB records")
+    parser.add_argument("path", type=Path, help="Path to LMDB directory")
+    parser.add_argument(
+        "keep",
+        help="Retention period, e.g. 14d or 24h. Use 'all' to delete every record",
+    )
+    args = parser.parse_args()
+
+    rotate_lmdb(args.path, args.keep)
+
+
+if __name__ == "__main__":
+    main()