Upload 12 files

app/services/adapters.py

@@ -2,7 +2,7 @@ import json
 from typing import Any, Protocol
 
 from app.models.workflow import (
-    CredentialPlaceholder,
+    CredentialReference,
     Position,
     WorkflowDocument,
     WorkflowEdge,
@@ -52,8 +52,8 @@ class N8nJsonAdapter:
             node_type = node.get("type", "n8n-nodes-base.noOp")
             position = node.get("position", [80 + index * 300, 200])
             credentials = {
-                key: CredentialPlaceholder(
-                    id=str(value.get("id", "__CREDENTIAL_ID__")),
+                key: CredentialReference(
+                    id=str(value["id"]) if value.get("id") else None,
                     name=str(value.get("name", f"Connect {key}")),
                     type=key,
                 )
@@ -140,7 +140,10 @@ class N8nJsonAdapter:
                     "position": [round(node.position.x), round(node.position.y)],
                     "parameters": node.data.parameters,
                     "credentials": {
-                        key: {"id": value.id, "name": value.name}
+                        key: {
+                            **({"id": value.id} if value.id else {}),
+                            "name": value.name,
+                        }
                         for key, value in (node.data.credentials or {}).items()
                     },
                     **({"disabled": True} if node.data.disabled else {}),

app/services/catalog.py

@@ -1,4 +1,4 @@
-from app.models.catalog import NodeDefinition, TemplateSummary
+from app.models.catalog import NodeDefinition
 
 
 NODE_CATALOG = [
@@ -102,7 +102,7 @@ NODE_CATALOG = [
         icon="Sparkles",
         color="#10a37f",
         typeVersion=1.8,
-        defaults={"modelId": "gpt-4.1-mini"},
+        defaults={"modelId": "gpt-5.5"},
         credentials=["openAiApi"],
     ),
     NodeDefinition(
@@ -129,42 +129,3 @@ NODE_CATALOG = [
 ]
 
 SUPPORTED_NODE_TYPES = {node.type for node in NODE_CATALOG}
-
-TEMPLATES = [
-    TemplateSummary(
-        id="invoice-intake",
-        name="AI invoice intake",
-        description="Extract invoice data from Gmail attachments and store it in Supabase.",
-        category="Finance",
-        node_count=6,
-        use_count=1842,
-        tags=["AI", "Gmail", "Supabase", "Slack"],
-    ),
-    TemplateSummary(
-        id="lead-enrichment",
-        name="Lead enrichment pipeline",
-        description="Enrich new CRM leads and route qualified accounts to sales.",
-        category="CRM",
-        node_count=9,
-        use_count=1270,
-        tags=["CRM", "AI", "Sales"],
-    ),
-    TemplateSummary(
-        id="support-triage",
-        name="Support ticket triage",
-        description="Classify, prioritize, and assign incoming support tickets.",
-        category="AI",
-        node_count=8,
-        use_count=966,
-        tags=["AI", "Support", "Slack"],
-    ),
-    TemplateSummary(
-        id="content-repurposing",
-        name="Content repurposing",
-        description="Turn long-form content into scheduled social posts.",
-        category="Social Media",
-        node_count=11,
-        use_count=2213,
-        tags=["Marketing", "AI", "Social"],
-    ),
-]

app/services/collaboration.py

@@ -6,6 +6,7 @@ from uuid import uuid4
 import httpx
 
 from app.core.config import Settings
+from app.core.errors import ServiceConfigurationError
 from app.models.workflow import (
     ShareRequest,
     ShareResponse,
@@ -16,9 +17,6 @@ from app.models.workflow import (
 )
 
 
-_demo_shares: dict[str, SharedWorkflowResponse] = {}
-
-
 class CollaborationRepository:
     def __init__(self, settings: Settings):
         self.settings = settings
@@ -39,6 +37,12 @@ class CollaborationRepository:
             "Prefer": prefer,
         }
 
+    def _require_configured(self) -> None:
+        if not self.configured:
+            raise ServiceConfigurationError(
+                "Supabase persistence is not configured on the API server."
+            )
+
     async def _workspace_access(
         self, client: httpx.AsyncClient, user_id: str, *, edit: bool
     ) -> set[str]:
@@ -99,6 +103,7 @@ class CollaborationRepository:
         return True
 
     async def create_share(self, request: ShareRequest, user_id: str) -> ShareResponse:
+        self._require_configured()
         token = secrets.token_urlsafe(32)
         token_hash = hashlib.sha256(token.encode()).hexdigest()
         expires_at = (
@@ -107,32 +112,23 @@ class CollaborationRepository:
             else None
         )
         share_id = str(uuid4())
-        if not self.configured:
-            if request.workflow is None:
-                raise ValueError("A workflow snapshot is required in local mode")
-            _demo_shares[token_hash] = SharedWorkflowResponse(
-                workflow=request.workflow,
-                permission=request.permission,
-                expires_at=expires_at,
+        async with httpx.AsyncClient(timeout=10) as client:
+            await self._authorized_workflow(
+                client, request.workflow_id, user_id, edit=True
             )
-        else:
-            async with httpx.AsyncClient(timeout=10) as client:
-                await self._authorized_workflow(
-                    client, request.workflow_id, user_id, edit=True
-                )
-                response = await client.post(
-                    f"{self.base_url}/workflow_shares",
-                    headers=self._headers(),
-                    json={
-                        "id": share_id,
-                        "workflow_id": request.workflow_id,
-                        "created_by": user_id,
-                        "token_hash": token_hash,
-                        "permission": request.permission,
-                        "expires_at": expires_at.isoformat() if expires_at else None,
-                    },
-                )
-                response.raise_for_status()
+            response = await client.post(
+                f"{self.base_url}/workflow_shares",
+                headers=self._headers(),
+                json={
+                    "id": share_id,
+                    "workflow_id": request.workflow_id,
+                    "created_by": user_id,
+                    "token_hash": token_hash,
+                    "permission": request.permission,
+                    "expires_at": expires_at.isoformat() if expires_at else None,
+                },
+            )
+            response.raise_for_status()
         return ShareResponse(
             id=share_id,
             url=f"{self.settings.allowed_origins[0].rstrip('/')}/share/{token}",
@@ -141,38 +137,33 @@ class CollaborationRepository:
         )
 
     async def shared_workflow(self, token: str) -> SharedWorkflowResponse:
+        self._require_configured()
         token_hash = hashlib.sha256(token.encode()).hexdigest()
-        if not self.configured:
-            shared = _demo_shares.get(token_hash)
-            if not shared:
+        async with httpx.AsyncClient(timeout=10) as client:
+            response = await client.get(
+                (
+                    f"{self.base_url}/workflow_shares"
+                    "?select=permission,expires_at,revoked_at,workflow:workflows(definition)"
+                    f"&token_hash=eq.{token_hash}&limit=1"
+                ),
+                headers=self._headers(),
+            )
+            response.raise_for_status()
+            rows = response.json()
+            if not rows or rows[0]["revoked_at"]:
                 raise ValueError("Share link was not found")
-        else:
-            async with httpx.AsyncClient(timeout=10) as client:
-                response = await client.get(
-                    (
-                        f"{self.base_url}/workflow_shares"
-                        "?select=permission,expires_at,revoked_at,workflow:workflows(definition)"
-                        f"&token_hash=eq.{token_hash}&limit=1"
-                    ),
-                    headers=self._headers(),
-                )
-                response.raise_for_status()
-                rows = response.json()
-                if not rows or rows[0]["revoked_at"]:
-                    raise ValueError("Share link was not found")
-                row = rows[0]
-                shared = SharedWorkflowResponse(
-                    workflow=WorkflowDocument.model_validate(row["workflow"]["definition"]),
-                    permission=row["permission"],
-                    expires_at=row["expires_at"],
-                )
+            row = rows[0]
+            shared = SharedWorkflowResponse(
+                workflow=WorkflowDocument.model_validate(row["workflow"]["definition"]),
+                permission=row["permission"],
+                expires_at=row["expires_at"],
+            )
         if shared.expires_at and shared.expires_at < datetime.now(UTC):
             raise ValueError("Share link has expired")
         return shared
 
     async def versions(self, workflow_id: str, user_id: str) -> list[VersionSummary]:
-        if not self.configured:
-            return []
+        self._require_configured()
         async with httpx.AsyncClient(timeout=10) as client:
             await self._authorized_workflow(client, workflow_id, user_id, edit=False)
             response = await client.get(
@@ -196,8 +187,7 @@ class CollaborationRepository:
         ]
 
     async def restore(self, workflow_id: str, version: int, user_id: str) -> WorkflowDocument:
-        if not self.configured:
-            raise ValueError("Version restore requires Supabase")
+        self._require_configured()
         async with httpx.AsyncClient(timeout=10) as client:
             await self._authorized_workflow(client, workflow_id, user_id, edit=True)
             response = await client.get(
@@ -233,8 +223,7 @@ class CollaborationRepository:
         return WorkflowDocument.model_validate(definition)
 
     async def comments(self, workflow_id: str, user_id: str) -> list[WorkflowComment]:
-        if not self.configured:
-            return []
+        self._require_configured()
         async with httpx.AsyncClient(timeout=10) as client:
             await self._authorized_workflow(client, workflow_id, user_id, edit=False)
             response = await client.get(
@@ -247,11 +236,7 @@ class CollaborationRepository:
     async def add_comment(
         self, workflow_id: str, body: str, node_id: str | None, user_id: str
     ) -> WorkflowComment:
-        if not self.configured:
-            return WorkflowComment(
-                id=str(uuid4()), workflow_id=workflow_id, user_id=user_id,
-                node_id=node_id, body=body, created_at=datetime.now(UTC).isoformat()
-            )
+        self._require_configured()
         async with httpx.AsyncClient(timeout=10) as client:
             await self._authorized_workflow(client, workflow_id, user_id, edit=False)
             response = await client.post(

app/services/generator.py

@@ -1,9 +1,13 @@
+import json
 import re
 from dataclasses import dataclass
 from typing import Protocol
 
+import httpx
+
+from app.core.config import get_settings
 from app.models.workflow import (
-    CredentialPlaceholder,
+    CredentialReference,
     GenerateWorkflowResponse,
     Position,
     WorkflowDocument,
@@ -13,9 +17,32 @@ from app.models.workflow import (
     WorkflowNodeData,
 )
 
+WORKFLOW_SYSTEM_PROMPT = (
+    "You generate production-ready n8n workflow documents for FlowForge. "
+    "Return only valid JSON with keys workflow, explanation, and warnings. "
+    "The workflow must contain name, active, nodes, edges, settings, and meta. "
+    "Each node must contain id, position {x,y}, and data with label, type, "
+    "typeVersion, category, parameters, and optional credentials. "
+    "Credential objects must include name and type and omit credential IDs. "
+    "Use only these categories: trigger, core, ai, database, communication, cloud, developer. "
+    "Use n8n expressions only when they start with = and have balanced {{ }} braces. "
+    "Never invent secret values, API keys, private URLs, or credential IDs. "
+    "For an unspecified endpoint use an n8n environment expression such as "
+    "={{ $env.SERVICE_BASE_URL }} and mention it in warnings."
+)
+
+
+def _parse_provider_response(content: str, provider: str, model: str) -> GenerateWorkflowResponse:
+    try:
+        result = GenerateWorkflowResponse.model_validate(json.loads(content))
+    except (TypeError, json.JSONDecodeError, ValueError) as exc:
+        raise ValueError(f"{provider} returned an invalid workflow document") from exc
+    result.workflow.meta.generatedBy = f"{provider} {model}"
+    return result
+
 
 class WorkflowProvider(Protocol):
-    async def generate(self, prompt: str) -> GenerateWorkflowResponse: ...
+    async def generate(self, prompt: str, model: str | None = None) -> GenerateWorkflowResponse: ...
 
 
 @dataclass(slots=True)
@@ -29,11 +56,11 @@ class PlannedNode:
     credentials: list[str]
 
 
-def _credential_map(types: list[str], label: str) -> dict[str, CredentialPlaceholder] | None:
+def _credential_map(types: list[str], label: str) -> dict[str, CredentialReference] | None:
     if not types:
         return None
     return {
-        credential_type: CredentialPlaceholder(
+        credential_type: CredentialReference(
             name=f"Connect {label}", type=credential_type
         )
         for credential_type in types
@@ -43,7 +70,7 @@ def _credential_map(types: list[str], label: str) -> dict[str, CredentialPlaceho
 class RuleBasedWorkflowProvider:
     """CPU-friendly baseline provider used as a fallback and in tests."""
 
-    async def generate(self, prompt: str) -> GenerateWorkflowResponse:
+    async def generate(self, prompt: str, model: str | None = None) -> GenerateWorkflowResponse:
         text = prompt.lower()
         plan: list[PlannedNode] = []
 
@@ -116,7 +143,7 @@ class RuleBasedWorkflowProvider:
                     "Call external API",
                     {
                         "method": "GET",
-                        "url": "https://api.example.com/resource",
+                        "url": "={{ $env.API_BASE_URL }}",
                         "options": {"timeout": 30000},
                     },
                     [],
@@ -241,7 +268,7 @@ class RuleBasedWorkflowProvider:
             },
             meta=WorkflowMeta(
                 description=prompt[:500],
-                generatedBy="FlowForge rule-based provider",
+                generatedBy="FlowForge deterministic provider",
                 version=1,
                 tags=["AI generated"],
             ),
@@ -250,12 +277,119 @@ class RuleBasedWorkflowProvider:
             workflow=workflow,
             explanation=f"Created a {len(nodes)}-node workflow with a trigger and connected actions.",
             warnings=[
-                "Replace credential placeholders before activating the workflow.",
+                "Select credentials for each connected service before activation.",
                 "Review generated expressions with representative execution data.",
             ],
         )
 
 
+class OpenAIWorkflowProvider:
+    """Generate normalized workflow documents through OpenAI structured JSON output."""
+
+    async def generate(self, prompt: str, model: str | None = None) -> GenerateWorkflowResponse:
+        settings = get_settings()
+        if not settings.openai_api_key:
+            raise RuntimeError("OPENAI_API_KEY is required when AI_PROVIDER=openai.")
+        selected_model = model or settings.openai_model
+        payload = {
+            "model": selected_model,
+            "messages": [
+                {"role": "system", "content": WORKFLOW_SYSTEM_PROMPT},
+                {"role": "user", "content": prompt},
+            ],
+            "response_format": {"type": "json_object"},
+            "max_completion_tokens": 12000,
+        }
+        async with httpx.AsyncClient(timeout=60) as client:
+            response = await client.post(
+                "https://api.openai.com/v1/chat/completions",
+                headers={
+                    "Authorization": f"Bearer {settings.openai_api_key}",
+                    "Content-Type": "application/json",
+                },
+                json=payload,
+            )
+            response.raise_for_status()
+        try:
+            content = response.json()["choices"][0]["message"]["content"]
+        except (KeyError, TypeError, IndexError) as exc:
+            raise ValueError("OpenAI returned an invalid response envelope") from exc
+        return _parse_provider_response(content, "OpenAI", selected_model)
+
+
+class GeminiWorkflowProvider:
+    """Generate workflow documents through the Gemini generateContent API."""
+
+    async def generate(self, prompt: str, model: str | None = None) -> GenerateWorkflowResponse:
+        settings = get_settings()
+        if not settings.gemini_api_key:
+            raise RuntimeError("GEMINI_API_KEY is required when AI_PROVIDER=gemini.")
+        selected_model = model or settings.gemini_model
+        payload = {
+            "systemInstruction": {"parts": [{"text": WORKFLOW_SYSTEM_PROMPT}]},
+            "contents": [{"role": "user", "parts": [{"text": prompt}]}],
+            "generationConfig": {
+                "responseMimeType": "application/json",
+                "temperature": 0.2,
+                "maxOutputTokens": 12000,
+            },
+        }
+        async with httpx.AsyncClient(timeout=60) as client:
+            response = await client.post(
+                f"https://generativelanguage.googleapis.com/v1beta/models/{selected_model}:generateContent",
+                headers={
+                    "x-goog-api-key": settings.gemini_api_key,
+                    "Content-Type": "application/json",
+                },
+                json=payload,
+            )
+            response.raise_for_status()
+        try:
+            parts = response.json()["candidates"][0]["content"]["parts"]
+            content = "".join(part.get("text", "") for part in parts)
+        except (KeyError, TypeError, IndexError) as exc:
+            raise ValueError("Gemini returned an invalid response envelope") from exc
+        return _parse_provider_response(content, "Gemini", selected_model)
+
+
+class OpenRouterWorkflowProvider:
+    """Generate workflow documents through OpenRouter's chat completions API."""
+
+    async def generate(self, prompt: str, model: str | None = None) -> GenerateWorkflowResponse:
+        settings = get_settings()
+        if not settings.openrouter_api_key:
+            raise RuntimeError(
+                "OPENROUTER_API_KEY is required when AI_PROVIDER=openrouter."
+            )
+        selected_model = model or settings.openrouter_model
+        payload = {
+            "model": selected_model,
+            "messages": [
+                {"role": "system", "content": WORKFLOW_SYSTEM_PROMPT},
+                {"role": "user", "content": prompt},
+            ],
+            "response_format": {"type": "json_object"},
+            "max_tokens": 12000,
+        }
+        async with httpx.AsyncClient(timeout=60) as client:
+            response = await client.post(
+                "https://openrouter.ai/api/v1/chat/completions",
+                headers={
+                    "Authorization": f"Bearer {settings.openrouter_api_key}",
+                    "Content-Type": "application/json",
+                    "HTTP-Referer": settings.allowed_origins[0],
+                    "X-Title": settings.app_name,
+                },
+                json=payload,
+            )
+            response.raise_for_status()
+        try:
+            content = response.json()["choices"][0]["message"]["content"]
+        except (KeyError, TypeError, IndexError) as exc:
+            raise ValueError("OpenRouter returned an invalid response envelope") from exc
+        return _parse_provider_response(content, "OpenRouter", selected_model)
+
+
 class ProviderRegistry:
     def __init__(self) -> None:
         self._providers: dict[str, WorkflowProvider] = {}
@@ -264,8 +398,14 @@ class ProviderRegistry:
         self._providers[name] = provider
 
     def get(self, name: str) -> WorkflowProvider:
-        return self._providers.get(name, self._providers["mock"])
+        provider = self._providers.get(name)
+        if not provider:
+            raise ValueError(f"Unsupported AI provider: {name}")
+        return provider
 
 
 providers = ProviderRegistry()
-providers.register("mock", RuleBasedWorkflowProvider())
+providers.register("openai", OpenAIWorkflowProvider())
+providers.register("gemini", GeminiWorkflowProvider())
+providers.register("openrouter", OpenRouterWorkflowProvider())
+providers.register("deterministic", RuleBasedWorkflowProvider())

app/services/repository.py

@@ -1,11 +1,20 @@
+import re
 from datetime import UTC, datetime
 from uuid import uuid4
 
 import httpx
 
 from app.core.config import Settings
-from app.models.catalog import ProjectSummary
-from app.models.workflow import SaveRequest, SaveResponse
+from app.core.errors import ServiceConfigurationError
+from app.models.catalog import (
+    DashboardStats,
+    DashboardSummary,
+    DashboardUser,
+    DashboardWorkflow,
+    ProjectSummary,
+    TemplateSummary,
+)
+from app.models.workflow import SaveRequest, SaveResponse, WorkflowDocument
 
 
 class WorkflowRepository:
@@ -24,13 +33,49 @@ class WorkflowRepository:
             "Prefer": "return=representation",
         }
 
+    def _require_configured(self) -> None:
+        if not self.configured:
+            raise ServiceConfigurationError(
+                "Supabase persistence is not configured on the API server."
+            )
+
+    @property
+    def base_url(self) -> str:
+        return f"{self.settings.supabase_url.rstrip('/')}/rest/v1"
+
+    async def _workspace_ids(self, client: httpx.AsyncClient, user_id: str) -> list[str]:
+        response = await client.get(
+            f"{self.base_url}/workspace_members?select=workspace_id&user_id=eq.{user_id}",
+            headers=self._headers(),
+        )
+        response.raise_for_status()
+        return [row["workspace_id"] for row in response.json()]
+
+    async def _count(self, client: httpx.AsyncClient, path: str) -> int:
+        response = await client.get(
+            f"{self.base_url}/{path}",
+            headers={**self._headers(), "Prefer": "count=exact"},
+        )
+        response.raise_for_status()
+        content_range = response.headers.get("content-range", "")
+        try:
+            return int(content_range.rsplit("/", 1)[1])
+        except (IndexError, ValueError):
+            return len(response.json())
+
+    @staticmethod
+    def _response_count(response: httpx.Response) -> int:
+        content_range = response.headers.get("content-range", "")
+        try:
+            return int(content_range.rsplit("/", 1)[1])
+        except (IndexError, ValueError):
+            return len(response.json())
+
     async def save(self, request: SaveRequest, user_id: str) -> SaveResponse:
+        self._require_configured()
         now = datetime.now(UTC).isoformat()
         workflow_id = str(uuid4())
-        if not self.configured:
-            return SaveResponse(id=workflow_id, version=1, saved_at=now)
-
-        base_url = f"{self.settings.supabase_url.rstrip('/')}/rest/v1"
+        base_url = self.base_url
         async with httpx.AsyncClient(timeout=10) as client:
             membership_response = await client.get(
                 (
@@ -128,17 +173,8 @@ class WorkflowRepository:
         )
 
     async def projects(self, user_id: str) -> list[ProjectSummary]:
-        if not self.configured:
-            return [
-                ProjectSummary(
-                    id="demo-finance",
-                    name="Finance Ops",
-                    description="Invoice and reporting automations",
-                    workflow_count=4,
-                    updated_at=datetime.now(UTC).isoformat(),
-                )
-            ]
-        base_url = f"{self.settings.supabase_url.rstrip('/')}/rest/v1"
+        self._require_configured()
+        base_url = self.base_url
         async with httpx.AsyncClient(timeout=10) as client:
             membership_response = await client.get(
                 f"{base_url}/workspace_members?select=workspace_id&user_id=eq.{user_id}",
@@ -161,4 +197,215 @@ class WorkflowRepository:
                 headers=self._headers(),
             )
             response.raise_for_status()
-        return [ProjectSummary(**item, workflow_count=0) for item in response.json()]
+            project_rows = response.json()
+            project_ids = [item["id"] for item in project_rows]
+            workflow_counts: dict[str, int] = {}
+            if project_ids:
+                project_filter = ",".join(project_ids)
+                workflow_response = await client.get(
+                    f"{base_url}/workflows?select=project_id&project_id=in.({project_filter})&is_archived=eq.false",
+                    headers=self._headers(),
+                )
+                workflow_response.raise_for_status()
+                for workflow in workflow_response.json():
+                    project_id = workflow.get("project_id")
+                    if project_id:
+                        workflow_counts[project_id] = workflow_counts.get(project_id, 0) + 1
+        return [
+            ProjectSummary(
+                **item,
+                workflow_count=workflow_counts.get(item["id"], 0),
+            )
+            for item in project_rows
+        ]
+
+    async def workflow(self, workflow_id: str, user_id: str) -> WorkflowDocument:
+        self._require_configured()
+        async with httpx.AsyncClient(timeout=10) as client:
+            workspace_ids = await self._workspace_ids(client, user_id)
+            response = await client.get(
+                f"{self.base_url}/workflows?select=workspace_id,definition&id=eq.{workflow_id}&is_archived=eq.false&limit=1",
+                headers=self._headers(),
+            )
+            response.raise_for_status()
+            rows = response.json()
+        if not rows or rows[0]["workspace_id"] not in workspace_ids:
+            raise ValueError("Workflow is not accessible by this user")
+        definition = dict(rows[0]["definition"])
+        definition["id"] = workflow_id
+        return WorkflowDocument.model_validate(definition)
+
+    async def template(self, template_id: str, user_id: str) -> WorkflowDocument:
+        self._require_configured()
+        async with httpx.AsyncClient(timeout=10) as client:
+            response = await client.get(
+                f"{self.base_url}/templates?select=name,definition,is_public,owner_id&id=eq.{template_id}&limit=1",
+                headers=self._headers(),
+            )
+            response.raise_for_status()
+            rows = response.json()
+        if not rows or (not rows[0]["is_public"] and rows[0]["owner_id"] != user_id):
+            raise ValueError("Template is not accessible by this user")
+        definition = dict(rows[0]["definition"])
+        definition.pop("id", None)
+        definition["name"] = rows[0]["name"]
+        return WorkflowDocument.model_validate(definition)
+
+    async def templates(
+        self,
+        user_id: str,
+        query: str = "",
+        category: str | None = None,
+        limit: int = 24,
+        offset: int = 0,
+    ) -> tuple[list[TemplateSummary], int]:
+        self._require_configured()
+        params: dict[str, str | int] = {
+            "select": "id,name,description,category,definition,use_count,tags",
+            "or": f"(is_public.eq.true,owner_id.eq.{user_id})",
+            "order": "use_count.desc",
+            "limit": limit,
+            "offset": offset,
+        }
+        if query:
+            safe_query = re.sub(r"[^\w\s-]", " ", query).strip()
+            if safe_query:
+                params["and"] = (
+                    f"(or(name.ilike.*{safe_query}*,description.ilike.*{safe_query}*))"
+                )
+        if category:
+            safe_category = re.sub(r"[^\w\s-]", "", category).strip()
+            if safe_category:
+                params["category"] = f"eq.{safe_category}"
+        async with httpx.AsyncClient(timeout=10) as client:
+            response = await client.get(
+                f"{self.base_url}/templates",
+                params=params,
+                headers={**self._headers(), "Prefer": "count=exact"},
+            )
+            response.raise_for_status()
+        items = [
+            TemplateSummary(
+                id=row["id"],
+                name=row["name"],
+                description=row.get("description") or "",
+                category=row["category"],
+                node_count=len((row.get("definition") or {}).get("nodes", [])),
+                use_count=row.get("use_count", 0),
+                tags=row.get("tags") or [],
+            )
+            for row in response.json()
+        ]
+        content_range = response.headers.get("content-range", "")
+        try:
+            total = int(content_range.rsplit("/", 1)[1])
+        except (IndexError, ValueError):
+            total = len(items)
+        return items, total
+
+    async def dashboard(self, user_id: str, email: str | None) -> DashboardSummary:
+        self._require_configured()
+        async with httpx.AsyncClient(timeout=15) as client:
+            workspace_ids = await self._workspace_ids(client, user_id)
+            profile_response = await client.get(
+                f"{self.base_url}/users?select=full_name,email,avatar_url&id=eq.{user_id}&limit=1",
+                headers=self._headers(),
+            )
+            profile_response.raise_for_status()
+            profile_rows = profile_response.json()
+            profile = profile_rows[0] if profile_rows else {}
+            display_name = profile.get("full_name") or (email or "Workspace member").split("@", 1)[0]
+
+            if not workspace_ids:
+                return DashboardSummary(
+                    workspace_name="No workspace",
+                    user=DashboardUser(
+                        name=display_name,
+                        email=profile.get("email") or email,
+                        avatar_url=profile.get("avatar_url"),
+                    ),
+                    stats=DashboardStats(),
+                )
+
+            workspace_filter = ",".join(workspace_ids)
+            workspace_response = await client.get(
+                f"{self.base_url}/workspaces?select=id,name&id=in.({workspace_filter})&order=created_at.asc",
+                headers=self._headers(),
+            )
+            project_response = await client.get(
+                f"{self.base_url}/projects?select=id,name&workspace_id=in.({workspace_filter})",
+                headers=self._headers(),
+            )
+            workflow_response = await client.get(
+                f"{self.base_url}/workflows?select=id,name,definition,is_active,updated_at,project_id&workspace_id=in.({workspace_filter})&is_archived=eq.false&order=updated_at.desc&limit=100",
+                headers={**self._headers(), "Prefer": "count=exact"},
+            )
+            favorite_response = await client.get(
+                f"{self.base_url}/favorites?select=workflow_id&user_id=eq.{user_id}&kind=eq.workflow",
+                headers=self._headers(),
+            )
+            for response in (
+                workspace_response,
+                project_response,
+                workflow_response,
+                favorite_response,
+            ):
+                response.raise_for_status()
+
+            projects = {row["id"]: row["name"] for row in project_response.json()}
+            favorites = {row["workflow_id"] for row in favorite_response.json()}
+            workflow_rows = workflow_response.json()
+            workflow_ids = [row["id"] for row in workflow_rows]
+            workflow_total = self._response_count(workflow_response)
+            active_workflows = await self._count(
+                client,
+                f"workflows?select=id&workspace_id=in.({workspace_filter})&is_archived=eq.false&is_active=eq.true&limit=1",
+            )
+            executions = successes = 0
+            if workflow_ids:
+                workflow_filter = ",".join(workflow_ids)
+                executions = await self._count(
+                    client,
+                    f"workflow_runs?select=id&workflow_id=in.({workflow_filter})&limit=1",
+                )
+                successes = await self._count(
+                    client,
+                    f"workflow_runs?select=id&workflow_id=in.({workflow_filter})&status=eq.success&limit=1",
+                )
+            ai_generations = await self._count(
+                client,
+                f"ai_history?select=id&user_id=eq.{user_id}&role=eq.assistant&limit=1",
+            )
+            template_items, _ = await self.templates(user_id, limit=6)
+
+        workspaces = workspace_response.json()
+        workspace_name = workspaces[0]["name"] if len(workspaces) == 1 else "All workspaces"
+        return DashboardSummary(
+            workspace_name=workspace_name,
+            user=DashboardUser(
+                name=display_name,
+                email=profile.get("email") or email,
+                avatar_url=profile.get("avatar_url"),
+            ),
+            stats=DashboardStats(
+                workflows=workflow_total,
+                projects=len(projects),
+                active_workflows=active_workflows,
+                executions=executions,
+                success_rate=round(successes / executions * 100, 1) if executions else None,
+                ai_generations=ai_generations,
+            ),
+            workflows=[
+                DashboardWorkflow(
+                    id=row["id"],
+                    name=row["name"],
+                    project_name=projects.get(row.get("project_id")),
+                    updated_at=row["updated_at"],
+                    node_count=len((row.get("definition") or {}).get("nodes", [])),
+                    is_active=bool(row["is_active"]),
+                    favorite=row["id"] in favorites,
+                )
+                for row in workflow_rows
+            ],
+            templates=template_items,
+        )

app/services/validation.py

@@ -37,7 +37,7 @@ class CredentialRule:
         issues = []
         for node in workflow.nodes:
             for credential in (node.data.credentials or {}).values():
-                if credential.id.startswith("__"):
+                if not credential.id:
                     issues.append(
                         ValidationIssue(
                             code=self.code,