gabriel / musehub public
musehub_domains.py python
311 lines 10.0 KB
Raw
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ breaking 144 days ago
1 """Domain plugin registry service — CRUD, manifest hashing, and discovery.
2
3 Provides all database operations for the musehub_domains and
4 musehub_domain_installs tables introduced in the V2 domain-agnostic migration.
5 """
6 from __future__ import annotations
7
8 import json
9 import uuid
10 from dataclasses import dataclass
11 from datetime import datetime, timezone
12
13 from sqlalchemy import func, select
14 from sqlalchemy.ext.asyncio import AsyncSession
15
16 from muse.core.types import blob_id
17 from musehub.core.genesis import compute_domain_id
18 from musehub.db.musehub_domain_models import MusehubDomain, MusehubDomainInstall
19 from musehub.db.utils import escape_like
20 from musehub.db.musehub_models import MusehubRepo
21 from musehub.types.json_types import JSONObject
22
23
24 def _utc_now() -> datetime:
25 return datetime.now(tz=timezone.utc)
26
27
28 def compute_manifest_hash(capabilities: JSONObject) -> str:
29 """Return the ``sha256:``-prefixed content ID of a capabilities JSON blob (sorted keys)."""
30 data = json.dumps(capabilities, sort_keys=True, separators=(",", ":")).encode()
31 return blob_id(data)
32
33
34 # ── Response dataclasses ──────────────────────────────────────────────────────
35
36
37 @dataclass
38 class DomainResponse:
39 domain_id: str
40 author_slug: str
41 slug: str
42 scoped_id: str # "@author/slug"
43 display_name: str
44 description: str
45 version: str
46 manifest_hash: str
47 capabilities: JSONObject
48 viewer_type: str
49 install_count: int
50 is_verified: bool
51 is_deprecated: bool
52 created_at: datetime
53 updated_at: datetime
54
55
56 @dataclass
57 class DomainListResponse:
58 domains: list[DomainResponse]
59 total: int
60 next_cursor: str | None = None
61
62
63 @dataclass
64 class DomainReposResponse:
65 domain_id: str
66 scoped_id: str
67 repos: list[JSONObject]
68 total: int
69 next_cursor: str | None = None
70
71
72 # ── Helpers ───────────────────────────────────────────────────────────────────
73
74
75 def _to_response(domain: MusehubDomain) -> DomainResponse:
76 return DomainResponse(
77 domain_id=domain.domain_id,
78 author_slug=domain.author_slug,
79 slug=domain.slug,
80 scoped_id=f"@{domain.author_slug}/{domain.slug}",
81 display_name=domain.display_name,
82 description=domain.description,
83 version=domain.version,
84 manifest_hash=domain.manifest_hash,
85 capabilities=dict(domain.capabilities) if domain.capabilities else {},
86 viewer_type=domain.viewer_type,
87 install_count=domain.install_count,
88 is_verified=domain.is_verified,
89 is_deprecated=domain.is_deprecated,
90 created_at=domain.created_at,
91 updated_at=domain.updated_at,
92 )
93
94
95 # ── Read operations ───────────────────────────────────────────────────────────
96
97
98 async def list_domains(
99 session: AsyncSession,
100 *,
101 query: str | None = None,
102 verified_only: bool = False,
103 cursor: str | None = None,
104 limit: int = 20,
105 ) -> DomainListResponse:
106 """List registered domains with optional text search and cursor-based pagination."""
107 stmt = select(MusehubDomain).where(MusehubDomain.is_deprecated.is_(False))
108
109 if verified_only:
110 stmt = stmt.where(MusehubDomain.is_verified.is_(True))
111
112 if query:
113 q = f"%{escape_like(query)}%"
114 stmt = stmt.where(
115 MusehubDomain.display_name.ilike(q, escape="\\")
116 | MusehubDomain.slug.ilike(q, escape="\\")
117 | MusehubDomain.author_slug.ilike(q, escape="\\")
118 | MusehubDomain.description.ilike(q, escape="\\")
119 )
120
121 count_stmt = select(func.count()).select_from(stmt.subquery())
122 total_result = await session.execute(count_stmt)
123 total = total_result.scalar_one()
124
125 stmt = stmt.order_by(MusehubDomain.install_count.desc(), MusehubDomain.created_at.desc())
126
127 # Apply cursor: filter rows where created_at < cursor_dt (DESC ordering)
128 # Normalize space→+ because URL-decoding can corrupt the ISO timezone offset (+00:00).
129 if cursor:
130 try:
131 cursor_dt = datetime.fromisoformat(cursor.replace(" ", "+"))
132 stmt = stmt.where(MusehubDomain.created_at < cursor_dt)
133 except ValueError:
134 pass # ignore malformed cursor, start from beginning
135
136 stmt = stmt.limit(limit + 1)
137 result = await session.execute(stmt)
138 domains = list(result.scalars().all())
139 has_more = len(domains) > limit
140 page_domains = domains[:limit]
141 next_cursor = page_domains[-1].created_at.isoformat() if (page_domains and has_more) else None
142
143 return DomainListResponse(
144 domains=[_to_response(d) for d in page_domains],
145 total=total,
146 next_cursor=next_cursor,
147 )
148
149
150 async def get_domain_by_scoped_id(
151 session: AsyncSession,
152 author_slug: str,
153 slug: str,
154 ) -> DomainResponse | None:
155 """Fetch a single domain by its @author/slug identity."""
156 stmt = select(MusehubDomain).where(
157 MusehubDomain.author_slug == author_slug,
158 MusehubDomain.slug == slug,
159 )
160 result = await session.execute(stmt)
161 domain = result.scalar_one_or_none()
162 return _to_response(domain) if domain else None
163
164
165 async def get_domain_by_id(
166 session: AsyncSession,
167 domain_id: str,
168 ) -> DomainResponse | None:
169 """Fetch a single domain by its UUID primary key."""
170 stmt = select(MusehubDomain).where(MusehubDomain.domain_id == domain_id)
171 result = await session.execute(stmt)
172 domain = result.scalar_one_or_none()
173 return _to_response(domain) if domain else None
174
175
176 async def list_repos_for_domain(
177 session: AsyncSession,
178 domain_id: str,
179 *,
180 cursor: str | None = None,
181 limit: int = 20,
182 ) -> DomainReposResponse:
183 """Return public repos using a specific domain plugin with cursor-based pagination."""
184 domain = await get_domain_by_id(session, domain_id)
185 if domain is None:
186 return DomainReposResponse(
187 domain_id=domain_id, scoped_id="", repos=[], total=0
188 )
189
190 base_where = (
191 MusehubRepo.domain_id == domain_id,
192 MusehubRepo.visibility == "public",
193 )
194 count_stmt = select(func.count()).select_from(
195 select(MusehubRepo).where(*base_where).subquery()
196 )
197 total_result = await session.execute(count_stmt)
198 total = total_result.scalar_one()
199
200 stmt = (
201 select(MusehubRepo)
202 .where(*base_where)
203 .order_by(MusehubRepo.created_at.desc())
204 )
205
206 # Apply cursor: filter rows where created_at < cursor_dt (DESC ordering)
207 # Normalize space→+ because URL-decoding can corrupt the ISO timezone offset (+00:00).
208 if cursor:
209 try:
210 cursor_dt = datetime.fromisoformat(cursor.replace(" ", "+"))
211 stmt = stmt.where(MusehubRepo.created_at < cursor_dt)
212 except ValueError:
213 pass # ignore malformed cursor, start from beginning
214
215 stmt = stmt.limit(limit + 1)
216 result = await session.execute(stmt)
217 repos = list(result.scalars().all())
218 has_more = len(repos) > limit
219 page_repos = repos[:limit]
220 next_cursor = page_repos[-1].created_at.isoformat() if (page_repos and has_more) else None
221
222 return DomainReposResponse(
223 domain_id=domain_id,
224 scoped_id=domain.scoped_id,
225 repos=[
226 {
227 "repo_id": r.repo_id,
228 "owner": r.owner,
229 "slug": r.slug,
230 "name": r.name,
231 "description": r.description,
232 "tags": list(r.tags) if r.tags else [],
233 "created_at": r.created_at.isoformat() if r.created_at else None,
234 }
235 for r in page_repos
236 ],
237 total=total,
238 next_cursor=next_cursor,
239 )
240
241
242 # ── Write operations ──────────────────────────────────────────────────────────
243
244
245 async def create_domain(
246 session: AsyncSession,
247 *,
248 author_user_id: str,
249 author_slug: str,
250 slug: str,
251 display_name: str,
252 description: str,
253 capabilities: JSONObject,
254 viewer_type: str = "generic",
255 version: str = "1.0.0",
256 ) -> DomainResponse:
257 """Register a new domain plugin in the MuseHub registry."""
258 manifest_hash = compute_manifest_hash(capabilities)
259 now = _utc_now()
260 domain = MusehubDomain(
261 domain_id=compute_domain_id(author_slug, slug, now.isoformat()),
262 author_user_id=author_user_id,
263 author_slug=author_slug,
264 slug=slug,
265 display_name=display_name,
266 description=description,
267 version=version,
268 manifest_hash=manifest_hash,
269 capabilities=capabilities,
270 viewer_type=viewer_type,
271 install_count=0,
272 is_verified=False,
273 is_deprecated=False,
274 created_at=now,
275 updated_at=now,
276 )
277 session.add(domain)
278 await session.flush()
279 return _to_response(domain)
280
281
282 async def record_domain_install(
283 session: AsyncSession,
284 user_id: str,
285 domain_id: str,
286 ) -> None:
287 """Record that a user has adopted a domain plugin (idempotent)."""
288 # Check if already installed
289 existing = await session.execute(
290 select(MusehubDomainInstall).where(
291 MusehubDomainInstall.user_id == user_id,
292 MusehubDomainInstall.domain_id == domain_id,
293 )
294 )
295 if existing.scalar_one_or_none() is not None:
296 return
297
298 install = MusehubDomainInstall(
299 install_id=str(uuid.uuid4()),
300 user_id=user_id,
301 domain_id=domain_id,
302 created_at=_utc_now(),
303 )
304 session.add(install)
305
306 # Increment install_count on the domain row
307 stmt = select(MusehubDomain).where(MusehubDomain.domain_id == domain_id)
308 result = await session.execute(stmt)
309 domain = result.scalar_one_or_none()
310 if domain is not None:
311 domain.install_count = (domain.install_count or 0) + 1
File History 1 commit
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ 144 days ago