Federation: Opensearch integration, federation init optimization, and P2P testing set-up

.dockerignore

@@ -0,0 +1,40 @@
+# Monorepo-root build context (gateway + federation Dockerfiles).
+# Keep gateway/ and federation/ and common/ available for COPY.
+
+**/.git
+**/.venv
+**/venv
+**/node_modules
+**/__pycache__
+**/.pytest_cache
+**/.ruff_cache
+**/dist
+**/build
+**/*.pyc
+
+# Secrets / local env
+**/.envs/
+federation-shared.env
+
+# Unreadable / bulky OpenSearch host data
+**/opensearch/data/
+
+# Other monorepo trees not needed for gateway/federation images
+sdk/
+jupyter/
+seaweedfs/
+docs/
+.cursor/
+
+# Gateway local noise (mirrored from gateway/.dockerignore intent)
+gateway/.editorconfig
+gateway/.gitattributes
+gateway/.github
+gateway/.idea
+gateway/.pre-commit-config.yaml
+gateway/.readthedocs.yaml
+gateway/.readthedocs.yml
+gateway/.travis.yaml
+gateway/.travis.yml
+gateway/.gitlab-ci.yaml
+gateway/.gitlab-ci.yml

.github/workflows/fed-code-quality.yaml

@@ -0,0 +1,105 @@
+# Workflow to run pre-commit checks and pytest for the Federation app
+# GitHub Action Workflow validator: https://rhysd.github.io/actionlint/
+name: federation-checks
+
+on:
+    workflow_dispatch:
+        # To manually trigger the workflow
+        # https://docs.github.com/en/actions/writing-workflows/choosing-when-your-workflow-runs/events-that-trigger-workflows#workflow_dispatch
+    push:
+        paths:
+            # https://docs.github.com/en/actions/writing-workflows/workflow-syntax-for-github-actions#filter-pattern-cheat-sheet
+            - federation/**
+            - common/** # Cover common/ changes too
+            - .pre-commit-config.yaml
+            - .github/workflows/fed-code-quality.yaml
+        branches:
+            - main
+            - master
+    pull_request:
+        paths:
+            # https://docs.github.com/en/actions/writing-workflows/workflow-syntax-for-github-actions#filter-pattern-cheat-sheet
+            - federation/**
+            - common/** # Cover common/ changes too
+            - .pre-commit-config.yaml
+            - .github/workflows/fed-code-quality.yaml
+        branches:
+            - main
+            - master
+        types:
+            # https://docs.github.com/en/actions/writing-workflows/choosing-when-your-workflow-runs/events-that-trigger-workflows#pull_request
+            - ready_for_review
+            - synchronize
+
+env:
+    DOCKER_BUILDKIT: 1
+    COMPOSE_DOCKER_CLI_BUILD: 1
+
+jobs:
+    # Federation hooks live in the repo-root .pre-commit-config.yaml;
+    # prek is a federation --extra dev dependency.
+    fed-pre-commit:
+        runs-on: ubuntu-latest
+        env:
+            UV_LINK_MODE: copy
+        steps:
+            - uses: actions/checkout@v7
+
+            - name: Install uv
+              uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
+              # https://github.com/marketplace/actions/astral-sh-setup-uv
+
+            - name: Cache prek hooks
+              id: cache-prek
+              uses: actions/cache@v6
+              # https://github.com/actions/cache/blob/main/examples.md#python---pip
+              with:
+                  key: prek-federation-${{ hashFiles('.pre-commit-config.yaml') }}
+                  path: ~/.cache/prek/
+
+            - name: Sync federation dependencies
+              working-directory: ./federation
+              # pyrefly-federation runs: cd federation && uv run --extra dev pyrefly check
+              run: uv sync --extra dev
+
+            - name: Install hooks
+              working-directory: ./federation
+              run: uv run --extra dev prek install --install-hooks
+
+            - name: Run federation prek hooks
+              working-directory: ./federation
+              # Only federation-scoped hooks (no biome / djLint / JS).
+              # prek walks up to the repo-root .pre-commit-config.yaml.
+              run: |
+                  uv run --extra dev prek run ruff-check-federation --all-files
+                  uv run --extra dev prek run ruff-format-federation --all-files
+                  uv run --extra dev prek run pyrefly-federation --all-files
+
+    # Run federation pytest (mocked deps — no Docker stack required)
+    fed-tests:
+        runs-on: ubuntu-latest
+        strategy:
+            matrix:
+                # uv will take care of installing other python versions,
+                # so we don't need a python-version matrix here.
+                platform: [ubuntu-latest]
+        steps:
+            - uses: actions/checkout@v7
+
+            - name: Install uv
+              uses: astral-sh/setup-uv@d31148d669074a8d0a63714ba94f3201e7020bc3 # v8.3.0
+              # https://github.com/marketplace/actions/astral-sh-setup-uv
+
+            - name: Install just on ubuntu
+              if: matrix.platform == 'ubuntu-latest'
+              working-directory: ./federation
+              run: |
+                  npm install -g rust-just
+
+            - name: Sync federation dependencies
+              working-directory: ./federation
+              run: uv sync --extra dev
+
+            - name: Run tests
+              working-directory: ./federation
+              run: just test

.github/workflows/gwy-code-quality.yaml

@@ -144,7 +144,7 @@ jobs:
             - name: Build and push
               uses: docker/build-push-action@v7
               with:
-                  context: gateway
+                  context: .
                   file: gateway/compose/production/django/Dockerfile
                   push: true
                   tags: ${{ steps.meta.outputs.tags }}

common/pyproject.toml

@@ -0,0 +1,18 @@
+[build-system]
+requires = ["uv_build>=0.11.24,<0.12"]
+build-backend = "uv_build"
+
+[tool.uv.build-backend]
+module-name = "sds_opensearch_query"
+module-root = ""
+
+[project]
+name = "sds-opensearch-query"
+version = "0.1.0"
+description = "Shared OpenSearch and federation helpers for SDS gateway and sync"
+requires-python = ">=3.13,<3.15"
+dependencies = [
+    "loguru>=0.7.2",
+    "opensearch-py>=2.7.1",
+    "requests>=2.32.0",
+]

common/sds_opensearch_query/__init__.py

@@ -0,0 +1,35 @@
+"""Framework-agnostic OpenSearch query helpers."""
+
+from sds_opensearch_query.filters import build_metadata_filter_clauses
+from sds_opensearch_query.filters import nested_query_clause
+from sds_opensearch_query.index_write import FED_CAPTURES_INDEX
+from sds_opensearch_query.index_write import FED_DATASETS_INDEX
+from sds_opensearch_query.index_write import federated_doc_id
+from sds_opensearch_query.index_write import index_federated_document
+from sds_opensearch_query.mapping import flatten_property_paths
+from sds_opensearch_query.query import bool_must_search_body
+from sds_opensearch_query.query import federation_not_deleted_clause
+from sds_opensearch_query.query import multi_match_clause
+from sds_opensearch_query.query import run_search
+from sds_opensearch_query.query import term_clause
+from sds_opensearch_query.redis_channel import FEDERATION_EVENTS_CHANNEL_PREFIX
+from sds_opensearch_query.redis_channel import federation_events_channel
+from sds_opensearch_query.redis_channel import resolve_federation_events_channel
+
+__all__ = [
+    "FEDERATION_EVENTS_CHANNEL_PREFIX",
+    "FED_CAPTURES_INDEX",
+    "FED_DATASETS_INDEX",
+    "bool_must_search_body",
+    "build_metadata_filter_clauses",
+    "federated_doc_id",
+    "federation_events_channel",
+    "federation_not_deleted_clause",
+    "flatten_property_paths",
+    "index_federated_document",
+    "multi_match_clause",
+    "nested_query_clause",
+    "resolve_federation_events_channel",
+    "run_search",
+    "term_clause",
+]

common/sds_opensearch_query/client.py

@@ -0,0 +1,35 @@
+from loguru import logger as log
+from opensearchpy import OpenSearch
+from opensearchpy import RequestsHttpConnection
+from requests.auth import HTTPBasicAuth
+
+
+def build_opensearch_client(
+    *,
+    host: str,
+    port: int,
+    user: str = "",
+    password: str = "",
+    use_ssl: bool = False,
+    verify_certs: bool = False,
+    ca_certs: str | None = None,
+) -> OpenSearch:
+    payload = {
+        "hosts": [{"host": host, "port": port}],
+        "use_ssl": use_ssl,
+        "verify_certs": verify_certs,
+        "ssl_show_warn": False,
+        "connection_class": RequestsHttpConnection,
+    }
+    if user:
+        payload["http_auth"] = HTTPBasicAuth(user, password)
+    if verify_certs:
+        if not ca_certs:
+            msg = (
+                "OPENSEARCH_VERIFY_CERTS is True but OPENSEARCH_CA_CERTS is not set. "
+                "Provide a CA bundle path or set OPENSEARCH_VERIFY_CERTS to False."
+            )
+            raise ValueError(msg)
+        log.info("OPENSEARCH_VERIFY_CERTS is True. Verifying certificates.")
+        payload["ca_certs"] = ca_certs
+    return OpenSearch(**payload)

common/sds_opensearch_query/filters.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from typing import Any
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+
+def nested_query_clause(
+    field_path: str,
+    query_type: str,
+    value: Any,
+    *,
+    levels_nested: int | None = None,
+    last_path: str | None = None,
+) -> dict[str, Any]:
+    """Build a nested OpenSearch clause for a dotted field path."""
+    if levels_nested is None:
+        levels_nested = field_path.count(".")
+
+    if levels_nested == 0:
+        key = f"{last_path}.{field_path}" if last_path else field_path
+        return {query_type: {key: value}}
+
+    path_parts = field_path.split(".")
+    current_path = path_parts[0]
+    if last_path is not None:
+        current_path = f"{last_path}.{current_path}"
+
+    return {
+        "nested": {
+            "path": current_path,
+            "query": nested_query_clause(
+                field_path=".".join(path_parts[1:]),
+                query_type=query_type,
+                value=value,
+                levels_nested=levels_nested - 1,
+                last_path=current_path,
+            ),
+        },
+    }
+
+
+def build_metadata_filter_clauses(
+    metadata_filters: list[dict[str, Any]] | None,
+    *,
+    known_field_paths: frozenset[str] | None = None,
+    on_unknown_field: Callable[[str], None] | None = None,
+) -> list[dict[str, Any]]:
+    """Turn API metadata filter dicts into OpenSearch query clauses."""
+    if not metadata_filters:
+        return []
+
+    clauses: list[dict[str, Any]] = []
+    for query in metadata_filters:
+        field_path: str = query["field_path"]
+        query_type: str = query["query_type"]
+        filter_value: Any = query["filter_value"]
+
+        if known_field_paths is not None and field_path not in known_field_paths:
+            if on_unknown_field is not None:
+                on_unknown_field(field_path)
+
+        levels_nested = field_path.count(".")
+        if levels_nested > 0:
+            clauses.append(
+                nested_query_clause(
+                    field_path=field_path,
+                    query_type=query_type,
+                    value=filter_value,
+                    levels_nested=levels_nested,
+                ),
+            )
+        else:
+            clauses.append({query_type: {field_path: filter_value}})
+
+    return clauses

common/sds_opensearch_query/index_write.py

@@ -0,0 +1,48 @@
+"""Shared fed-* OpenSearch document id and index write helpers."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from typing import Any
+
+if TYPE_CHECKING:
+    from datetime import datetime
+    from uuid import UUID
+
+    from opensearchpy import OpenSearch
+
+FED_DATASETS_INDEX = "fed-datasets"
+FED_CAPTURES_INDEX = "fed-captures"
+
+
+def federated_doc_id(site_name: str, uuid: UUID | str) -> str:
+    """Stable OpenSearch ``_id`` for a site-owned federated asset."""
+    return f"{site_name}:{uuid}"
+
+
+def index_federated_document(
+    client: OpenSearch,
+    *,
+    index_name: str,
+    site_name: str,
+    uuid: UUID,
+    body: dict[str, Any],
+    event_at: datetime,
+    refresh: str | bool = "wait_for",
+) -> str:
+    """Index a federated document and stamp ``federation_event_at``.
+
+    Returns the document id written.
+    """
+    doc_id = federated_doc_id(site_name, uuid)
+    doc = {
+        **body,
+        "federation_event_at": event_at.isoformat(),
+    }
+    client.index(
+        index=index_name,
+        id=doc_id,
+        body=doc,
+        refresh=refresh,
+    )
+    return doc_id