import hashlib

from typing import Any, Dict, List, Optional

from feast.errors import FeatureViewNotFoundException

from feast.feature_view import FeatureView

def feature_view_to_vs_id(project: str, feature_view_name: str) -> str:

digest = hashlib.sha256(f"{project}:{feature_view_name}".encode()).hexdigest()[:24]

return f"vs_{digest}"

def _has_vector_index(fv: FeatureView) -> bool:

return any(getattr(f, "vector_index", False) for f in fv.schema)

def build_vector_store_object(project: str, fv: FeatureView) -> Dict[str, Any]:

return {

"id": feature_view_to_vs_id(project, fv.name),

"object": "vector_store",

"name": fv.name,

"status": "completed",

"created_at": int(fv.created_timestamp.timestamp())

if fv.created_timestamp

else 0,

}

class VectorStoreRegistry:

"""Cached mapping from vs_{hash} IDs to FeatureViews.

Built once at server startup, refreshed on registry TTL cycle.

"""

def __init__(self, store: Any):

self._store = store

self._id_to_fv: Dict[str, FeatureView] = {}

self._vector_store_objects: List[Dict[str, Any]] = []

self.refresh()

def refresh(self) -> None:

project = self._store.project

id_to_fv: Dict[str, FeatureView] = {}

objects: List[Dict[str, Any]] = []

for fv in self._store.list_feature_views():

if _has_vector_index(fv):

vs_id = feature_view_to_vs_id(project, fv.name)

id_to_fv[vs_id] = fv

objects.append(build_vector_store_object(project, fv))

self._id_to_fv = id_to_fv

self._vector_store_objects = objects

def resolve(self, vs_id: str) -> FeatureView:

fv = self._id_to_fv.get(vs_id)

if fv is None:

raise FeatureViewNotFoundException(vs_id)

return fv

def list_vector_stores(self) -> List[Dict[str, Any]]:

return self._vector_store_objects

def get_vector_store(self, vs_id: str) -> Optional[Dict[str, Any]]:

fv = self._id_to_fv.get(vs_id)

if fv is None:

return None

return build_vector_store_object(self._store.project, fv)