"""Catalog storage for skills and protocols (PostgreSQL).
One table holds both kinds; ownership is expressed by ``owner_id`` plus
``visibility``:
- personal rows have ``owner_id = <user id>`` and only that user sees them;
- shared rows have ``owner_id = ''`` and everybody sees them, only an admin
writes them.
Every write appends to ``skill_versions``, so the history is the audit log of
who changed a procedure and why. ``content_hash`` is what a hive-sync service
will compare later; nothing else about synchronization is decided yet.
"""
from __future__ import annotations
import asyncio
import json
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any
import asyncpg
import structlog
from navi.skills.models import (
KINDS,
VISIBILITIES,
SkillDoc,
SkillError,
content_hash,
parse_document,
)
log = structlog.get_logger()
_DDL = """
CREATE TABLE IF NOT EXISTS skills (
id TEXT PRIMARY KEY,
kind TEXT NOT NULL CHECK (kind IN ('skill','protocol')),
name TEXT NOT NULL,
description TEXT NOT NULL DEFAULT '',
triggers TEXT[] NOT NULL DEFAULT '{}',
tags TEXT[] NOT NULL DEFAULT '{}',
area TEXT NOT NULL DEFAULT 'general',
visibility TEXT NOT NULL DEFAULT 'personal'
CHECK (visibility IN ('personal','shared')),
owner_id TEXT NOT NULL DEFAULT '',
content TEXT NOT NULL,
frontmatter JSONB NOT NULL DEFAULT '{}'::jsonb,
version INTEGER NOT NULL DEFAULT 1,
content_hash TEXT NOT NULL DEFAULT '',
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_by TEXT NOT NULL DEFAULT ''
);
CREATE UNIQUE INDEX IF NOT EXISTS uq_skills_owner_name
ON skills (owner_id, kind, name);
CREATE INDEX IF NOT EXISTS idx_skills_visible
ON skills (visibility, kind, updated_at DESC);
CREATE TABLE IF NOT EXISTS skill_versions (
id BIGSERIAL PRIMARY KEY,
skill_id TEXT NOT NULL REFERENCES skills(id) ON DELETE CASCADE,
version INTEGER NOT NULL,
content TEXT NOT NULL,
frontmatter JSONB NOT NULL DEFAULT '{}'::jsonb,
content_hash TEXT NOT NULL DEFAULT '',
edited_by TEXT NOT NULL DEFAULT '',
reason TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
UNIQUE (skill_id, version)
);
"""
_SELECT = (
"id, kind, name, description, triggers, tags, area, visibility, owner_id, "
"content, version, content_hash, created_at, updated_at, updated_by"
)
@dataclass
class SkillEntry:
id: str
kind: str
name: str
description: str
triggers: list[str]
tags: list[str]
area: str
visibility: str
owner_id: str
content: str
version: int
content_hash: str
created_at: datetime | None = None
updated_at: datetime | None = None
updated_by: str = ""
@property
def doc(self) -> SkillDoc:
return parse_document(self.content, kind=self.kind)
def to_dict(self, *, include_content: bool = True) -> dict[str, Any]:
data = {
"id": self.id,
"kind": self.kind,
"name": self.name,
"description": self.description,
"triggers": self.triggers,
"tags": self.tags,
"area": self.area,
"visibility": self.visibility,
"owner_id": self.owner_id,
"version": self.version,
"content_hash": self.content_hash,
"created_at": self.created_at.isoformat() if self.created_at else None,
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
"updated_by": self.updated_by,
}
if include_content:
data["content"] = self.content
return data
def _row_to_entry(row) -> SkillEntry:
return SkillEntry(
id=row["id"],
kind=row["kind"],
name=row["name"],
description=row["description"] or "",
triggers=list(row["triggers"] or []),
tags=list(row["tags"] or []),
area=row["area"] or "general",
visibility=row["visibility"],
owner_id=row["owner_id"] or "",
content=row["content"],
version=row["version"],
content_hash=row["content_hash"] or "",
created_at=row["created_at"],
updated_at=row["updated_at"],
updated_by=row["updated_by"] or "",
)
def _json(frontmatter: dict[str, Any]) -> str:
return json.dumps(frontmatter, ensure_ascii=False)
class SkillStore:
def __init__(self, pool: asyncpg.Pool) -> None:
self._pool = pool
self._ready = False
self._lock = asyncio.Lock()
async def _ensure_ready(self) -> None:
if self._ready:
return
async with self._lock:
if self._ready:
return
async with self._pool.acquire() as conn:
await conn.execute(_DDL)
self._ready = True
# ── reads ───────────────────────────────────────────────────────────────
async def get(self, skill_id: str) -> SkillEntry | None:
await self._ensure_ready()
async with self._pool.acquire() as conn:
row = await conn.fetchrow(
f"SELECT {_SELECT} FROM skills WHERE id = $1", skill_id
)
return _row_to_entry(row) if row else None
async def find(
self, *, user_id: str, name: str | None = None, kind: str | None = None,
skill_id: str | None = None,
) -> SkillEntry | None:
"""One entry by id or by name, within what ``user_id`` may see."""
await self._ensure_ready()
clauses = ["(owner_id = $1 OR visibility = 'shared')"]
args: list[Any] = [user_id]
if skill_id:
args.append(skill_id)
clauses.append(f"id = ${len(args)}")
if name:
args.append(name)
clauses.append(f"lower(name) = lower(${len(args)})")
if kind:
args.append(kind)
clauses.append(f"kind = ${len(args)}")
async with self._pool.acquire() as conn:
row = await conn.fetchrow(
f"SELECT {_SELECT} FROM skills WHERE " + " AND ".join(clauses) +
" ORDER BY (owner_id = $1) DESC LIMIT 1",
*args,
)
return _row_to_entry(row) if row else None
async def list(
self,
*,
user_id: str,
kind: str | None = None,
query: str | None = None,
tag: str | None = None,
area: str | None = None,
limit: int = 100,
) -> list[SkillEntry]:
await self._ensure_ready()
clauses = ["(owner_id = $1 OR visibility = 'shared')"]
args: list[Any] = [user_id]
if kind:
args.append(kind)
clauses.append(f"kind = ${len(args)}")
if query:
args.append(f"%{query}%")
n = len(args)
clauses.append(
f"(name ILIKE ${n} OR description ILIKE ${n} "
f"OR array_to_string(triggers, ' ') ILIKE ${n} "
f"OR array_to_string(tags, ' ') ILIKE ${n})"
)
if tag:
args.append(tag)
clauses.append(f"${len(args)} = ANY(tags)")
if area:
args.append(area)
clauses.append(f"area = ${len(args)}")
args.append(max(1, min(int(limit), 500)))
async with self._pool.acquire() as conn:
rows = await conn.fetch(
f"SELECT {_SELECT} FROM skills WHERE " + " AND ".join(clauses) +
f" ORDER BY (owner_id = $1) DESC, updated_at DESC LIMIT ${len(args)}",
*args,
)
return [_row_to_entry(r) for r in rows]
async def list_versions(self, skill_id: str, limit: int = 50) -> list[dict]:
await self._ensure_ready()
async with self._pool.acquire() as conn:
rows = await conn.fetch(
"SELECT version, content, edited_by, reason, created_at, content_hash "
"FROM skill_versions WHERE skill_id = $1 "
"ORDER BY version DESC LIMIT $2",
skill_id, max(1, min(int(limit), 200)),
)
return [
{
"version": r["version"],
"content": r["content"],
"edited_by": r["edited_by"],
"reason": r["reason"],
"created_at": r["created_at"].isoformat() if r["created_at"] else None,
"content_hash": r["content_hash"],
}
for r in rows
]
# ── writes ──────────────────────────────────────────────────────────────
async def create(
self,
*,
owner_id: str,
content: str,
visibility: str = "personal",
edited_by: str = "",
reason: str | None = None,
kind: str | None = None,
) -> SkillEntry:
await self._ensure_ready()
if visibility not in VISIBILITIES:
raise SkillError(f"visibility must be one of {', '.join(VISIBILITIES)}")
doc = parse_document(content, kind=kind)
if visibility == "shared":
owner_id = ""
skill_id = uuid.uuid4().hex
digest = content_hash(content)
try:
async with self._pool.acquire() as conn:
async with conn.transaction():
await conn.execute(
"""
INSERT INTO skills (id, kind, name, description, triggers, tags,
area, visibility, owner_id, content, frontmatter,
version, content_hash, updated_by)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11::jsonb,
1, $12, $13)
""",
skill_id, doc.kind, doc.name, doc.description,
doc.triggers, doc.tags, doc.area, visibility, owner_id,
content, _json(doc.frontmatter()), digest, edited_by,
)
await conn.execute(
"""
INSERT INTO skill_versions (skill_id, version, content, frontmatter,
content_hash, edited_by, reason)
VALUES ($1, 1, $2, $3::jsonb, $4, $5, $6)
""",
skill_id, content, _json(doc.frontmatter()), digest, edited_by, reason,
)
except asyncpg.UniqueViolationError as e:
raise SkillError(
f"a {doc.kind} named {doc.name!r} already exists in this scope"
) from e
entry = await self.get(skill_id)
assert entry is not None
log.info(
"skills.created", skill_id=skill_id, kind=doc.kind, name=doc.name,
visibility=visibility, owner_id=owner_id,
)
return entry
async def update(
self,
skill_id: str,
*,
content: str,
edited_by: str = "",
reason: str | None = None,
) -> SkillEntry | None:
await self._ensure_ready()
current = await self.get(skill_id)
if current is None:
return None
doc = parse_document(content, kind=current.kind)
digest = content_hash(content)
new_version = current.version + 1
try:
async with self._pool.acquire() as conn:
async with conn.transaction():
await conn.execute(
"""
INSERT INTO skill_versions (skill_id, version, content, frontmatter,
content_hash, edited_by, reason)
VALUES ($1, $2, $3, $4::jsonb, $5, $6, $7)
""",
skill_id, new_version, content, _json(doc.frontmatter()),
digest, edited_by, reason,
)
await conn.execute(
"""
UPDATE skills SET name = $2, description = $3, triggers = $4,
tags = $5, area = $6, content = $7, frontmatter = $8::jsonb,
version = $9, content_hash = $10, updated_at = now(),
updated_by = $11
WHERE id = $1
""",
skill_id, doc.name, doc.description, doc.triggers, doc.tags,
doc.area, content, _json(doc.frontmatter()), new_version,
digest, edited_by,
)
except asyncpg.UniqueViolationError as e:
raise SkillError(
f"a {doc.kind} named {doc.name!r} already exists in this scope"
) from e
log.info("skills.updated", skill_id=skill_id, version=new_version, edited_by=edited_by)
return await self.get(skill_id)
async def delete(self, skill_id: str) -> bool:
await self._ensure_ready()
async with self._pool.acquire() as conn:
status = await conn.execute("DELETE FROM skills WHERE id = $1", skill_id)
return status.endswith(" 1")
async def set_visibility(
self, skill_id: str, visibility: str, *, edited_by: str = ""
) -> SkillEntry | None:
"""Move an entry between personal and shared.
Promoting a personal entry moves its owner to '' (shared); demoting
requires an owner, so the caller passes ``edited_by`` as the new owner.
"""
await self._ensure_ready()
if visibility not in VISIBILITIES:
raise SkillError(f"visibility must be one of {', '.join(VISIBILITIES)}")
owner_id = "" if visibility == "shared" else edited_by
try:
async with self._pool.acquire() as conn:
await conn.execute(
"UPDATE skills SET visibility = $2, owner_id = $3, updated_at = now(), "
"updated_by = $4 WHERE id = $1",
skill_id, visibility, owner_id, edited_by,
)
except asyncpg.UniqueViolationError as e:
raise SkillError("an entry with this name already exists in the target scope") from e
return await self.get(skill_id)
def is_kind(value: str) -> bool:
return value in KINDS
@dataclass
class SkillRef:
"""A compact catalog row for search results."""
id: str
kind: str
name: str
description: str
tags: list[str] = field(default_factory=list)
area: str = "general"
visibility: str = "personal"
version: int = 1
@classmethod
def of(cls, entry: SkillEntry) -> "SkillRef":
return cls(
id=entry.id, kind=entry.kind, name=entry.name,
description=entry.description, tags=entry.tags, area=entry.area,
visibility=entry.visibility, version=entry.version,
)
def line(self) -> str:
bits = [f"[{self.kind}]", self.name]
if self.version and self.version > 1:
bits.append(f"v{self.version}")
bits.append(f"({self.area}/{self.visibility})")
line = " ".join(bits)
if self.description:
line += f" — {self.description}"
if self.tags:
line += f" [tags: {', '.join(self.tags)}]"
line += f" id={self.id}"
return line
