RankFlow AI — Refined Technical Specification

Agent-Centric, Skill-Based Local SEO Automation Platform#

Version: 2.0.0
Date: 2025
Status: Architecture Blueprint — Agent-Buildable
Infrastructure: AWS + Dokploy
Philosophy: Skills are the single source of truth. Code is generated from skills. Runtime behavior is driven by skills.

---

1. Core Philosophy#

1.1 Skills Are the Source of Truth#

Every capability of RankFlow is defined by a skill file. Skills are markdown documents with structured frontmatter that serve as:

• Declarative specifications — what the feature does, inputs, outputs, errors
• Implementation guides — how coding agents should build it
• Runtime contracts — how the harness executes and validates the feature
• Observability schemas — what to log, measure, and alert on

Principle: If you want to change how something works, you modify the skill. A coding agent reads the skill and updates the code. A runtime agent reads the skill and executes the workflow.

1.2 Agent-Centric Design#

The platform is built by agents, for agents:

• Coding agents read skills → generate/modify code → test against skill acceptance criteria
• Runtime agents (the harness) read skills → plan execution → dispatch workflows → observe results
• Self-healing: When a skill fails, the harness logs the failure, attempts fallback skills, and escalates to human if needed

1.3 Deterministic Workflows#

Every business process is a durable workflow managed by Restate:

• Workflows survive crashes, restarts, and deployments
• Steps are retried with exponential backoff
• External events (OAuth callbacks, webhooks) are awaited natively
• State is transparent — you can inspect exactly where any workflow is

1.4 API-First Everything#

No local binaries, no Puppeteer on the server, no Chrome in Docker:

• Browser automation → Hyperbrowser API, Firecrawl API, ScrapeGraph API
• Social media → Composio API, Zernio API, direct platform APIs
• SEO data → DataForSEO API
• AI → Claude/GPT-4o APIs

1.5 Lean and Observable#

• One deploy target: AWS + Dokploy (not Vercel + VPS split)
• One queue system: Redis Streams (not BullMQ)
• One workflow engine: Restate (not custom job tables)
• One logger: Structured JSON logs (Pino + Python structlog)
• One schema: ~15 core models (not 30+)

---

2. Infrastructure (AWS + Dokploy)#

2.1 Why Dokploy?#

Dokploy is a self-hosted PaaS that runs on your AWS infrastructure. It gives you:

• Heroku-like experience on your own servers
• Git-based deployments with auto-build
• Database provisioning (Postgres, Redis, MySQL)
• SSL certificates (Let's Encrypt)
• Environment variables management
• Monitoring (basic logs, resource usage)
• Multi-server support (scale later)

You own the infrastructure. No Vercel lock-in. No surprise bills.

2.2 AWS Architecture#

AWS Account (ap-south-1 / Mumbai)
│
├── VPC
│   ├── Public Subnet
│   │   └── EC2 (t3.large / t3.xlarge) — Dokploy + Apps
│   ├── Private Subnet (future)
│   │   └── RDS PostgreSQL (future — start with Dokploy Postgres)
│   └── Security Groups
│
├── S3 Bucket
│   ├── assets.rankflow.ai/     # Doctor photos, logos
│   ├── reports.rankflow.ai/    # Generated PDFs (if needed later)
│   └── backups.rankflow.ai/    # DB backups
│
├── Route 53
│   ├── rankflow.ai             # Main domain
│   ├── *.rankflow.ai           # Wildcard for subdomains
│   └── api.rankflow.ai         # API subdomain
│
├── CloudFront (optional, future)
│   └── CDN for static assets
│
└── IAM Roles
    ├── DokployEC2Role          # S3 access, CloudWatch logs
    └── RestateRole             # If using Restate Cloud

2.3 Dokploy Services#

Total estimated cost: ~$60-80/month for the EC2 instance + Dokploy services.

2.4 Environment Variables (Dokploy)#

# App
APP_ENV=production
APP_URL=https://rankflow.ai
API_URL=https://api.rankflow.ai

# Database
DATABASE_URL=postgresql://rankflow:xxx@rankflow-db:5432/rankflow

# Redis
REDIS_URL=redis://rankflow-cache:6379

# Auth (Better Auth)
BETTER_AUTH_SECRET=xxx
BETTER_AUTH_URL=https://rankflow.ai

# Google OAuth + GBP
GOOGLE_CLIENT_ID=xxx
GOOGLE_CLIENT_SECRET=xxx

# AI
ANTHROPIC_API_KEY=xxx
OPENAI_API_KEY=xxx

# SEO Data
DATAFORSEO_LOGIN=xxx
DATAFORSEO_PASSWORD=xxx

# Browser Automation
HYPERBROWSER_API_KEY=xxx
FIRECRAWL_API_KEY=xxx
SCRAPEGRAPH_API_KEY=xxx

# Social
COMPOSIO_API_KEY=xxx
ZERNIO_API_KEY=xxx

# Storage (S3)
AWS_ACCESS_KEY_ID=xxx
AWS_SECRET_ACCESS_KEY=xxx
AWS_S3_BUCKET=rankflow-assets
AWS_REGION=ap-south-1

# Restate
RESTATE_ENDPOINT=http://rankflow-worker:8080

# Email
RESEND_API_KEY=xxx

# Monitoring
SENTRY_DSN=xxx

---

3. System Architecture#

3.1 High-Level Diagram#

┌─────────────────────────────────────────────────────────────────────┐
│                         CLIENT LAYER                                │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  ┌──────────┐  │
│  │   Doctor     │  │   Doctor     │  │   Public     │  │  Admin   │  │
│  │  Dashboard   │  │   Site       │  │   Site       │  │  Panel   │  │
│  │  (Next.js)   │  │  (Next.js)   │  │  (Next.js)   │  │(Next.js) │  │
│  └──────────────┘  └──────────────┘  └──────────────┘  └──────────┘  │
└─────────────────────────────────────────────────────────────────────┘
                              │
                              ▼ HTTP / WebSocket
┌─────────────────────────────────────────────────────────────────────┐
│                         API GATEWAY (FastAPI)                         │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐              │
│  │   REST API   │  │   Webhooks   │  │   Health     │              │
│  │   (Routers)  │  │  (Stripe,    │  │   /metrics   │              │
│  │              │  │  Composio,   │  │              │              │
│  │              │  │  Google)     │  │              │              │
│  └──────────────┘  └──────────────┘  └──────────────┘              │
│                                                                     │
│  Middleware: Auth (Better Auth), Rate Limit, Request ID, Logging    │
└─────────────────────────────────────────────────────────────────────┘
                              │
                              ▼ Internal API
┌─────────────────────────────────────────────────────────────────────┐
│                      SKILL HARNESS (Python)                         │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐              │
│  │   Skill      │  │   Workflow   │  │   State      │              │
│  │   Registry   │  │   Planner    │  │   Manager    │              │
│  │  (loads .md) │  │  (plans exec)│  │  (Restate)   │              │
│  └──────────────┘  └──────────────┘  └──────────────┘              │
│                                                                     │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐              │
│  │   Execution  │  │   Fallback │  │   Audit      │              │
│  │   Engine     │  │   Handler    │  │   Logger     │              │
│  │  (dispatches)│  │  (retries)   │  │  (structured)│              │
│  └──────────────┘  └──────────────┘  └──────────────┘              │
└─────────────────────────────────────────────────────────────────────┘
                              │
                              ▼ API Calls
┌─────────────────────────────────────────────────────────────────────┐
│                      EXTERNAL SERVICES                              │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│  │  Claude  │ │Hyper-    │ │DataForSEO│ │ Composio │ │  Resend  │ │
│  │  Sonnet  │ │browser  │ │          │ │          │ │          │ │
│  │  GPT-4o  │ │Firecrawl │ │          │ │  Zernio  │ │   S3     │ │
│  │  (AI)    │ │(Browser) │ │  (SEO)   │ │ (Social) │ │(Storage) │ │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────────┐
│                         DATA LAYER                                    │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐          │
│  │PostgreSQL│  │  Redis   │  │   S3     │  │  Restate │          │
│  │ (Prisma) │  │(Streams,│  │ (Assets) │  │  State   │          │
│  │          │  │  Cache)  │  │          │  │  Store   │          │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘          │
└─────────────────────────────────────────────────────────────────────┘

3.2 Request Flow#

1. Doctor clicks "Generate GBP Post" in Next.js dashboard
2. Next.js calls FastAPI /api/v1/skills/gbp-post-create with practice_id, content
3. FastAPI validates auth via Better Auth session
4. FastAPI calls Harness execute_skill("gbp-post-create", payload)
5. Harness loads skill skills/04-gbp-post-create.md
6. Harness plans execution: validate → generate → publish → log
7. Harness dispatches to Restate workflow GbpPostCreateWorkflow
8. Restate executes steps with automatic retries and state persistence
9. Restate calls external APIs (Claude for content, GBP API for publish)
10. Restate stores result in Postgres, updates job status
11. Harness logs structured audit entry
12. FastAPI returns result to Next.js
13. Next.js shows success + audit trail

---

4. The Skill System#

4.1 Skill File Format#

Every skill is a markdown file with YAML frontmatter:

---
id: 04-gbp-post-create
name: Create GBP Post
category: gbp
version: 1.0.0
autonomous: true              # Can run without human approval
approval_required: false      # If false, runs immediately
triggers:
  - manual
  - scheduled
  - ai-generated
inputs:
  practice_id:
    type: string
    required: true
    description: "The practice ID"
  location_id:
    type: string
    required: true
    description: "The GBP location ID"
  content:
    type: string
    required: false
    description: "Post content (if not provided, AI generates)"
    max_length: 1500
  media_urls:
    type: array[string]
    required: false
    description: "URLs of images/videos to attach"
    max_items: 10
  cta_type:
    type: enum[BOOK, CALL, LEARN_MORE, SIGN_UP, ORDER]
    required: false
    default: BOOK
outputs:
  post_id:
    type: string
    description: "The created post ID"
  status:
    type: enum[SCHEDULED, PUBLISHED, FAILED]
  published_url:
    type: string
    required: false
dependencies:
  - 03-gbp-connect
  - 21-content-generate
external_apis:
  - google-business-profile
  - claude-sonnet
cost_estimate_usd: 0.05
time_estimate_seconds: 10
retry_policy:
  max_attempts: 3
  backoff: exponential
  initial_delay_seconds: 5
observability:
  metrics:
    - gbp_posts_created_total
    - gbp_post_create_duration_seconds
    - gbp_post_create_cost_usd
  alerts:
    - condition: "status == FAILED"
      severity: warning
      message: "GBP post creation failed"
    - condition: "duration_seconds > 30"
      severity: warning
      message: "GBP post creation slow"
---

## Purpose
Create and publish a Google Business Profile post for a medical practice.

## Business Context
GBP posts appear in Google Search and Maps results. They drive patient engagement
and improve local SEO. Medical practices should post 1-2x per week.

## Procedure

### Step 1: Validate Prerequisites
- Check practice has active GBP connection (skill 03)
- Check GBP API quota (max 500 posts/day per location)
- Verify location_id belongs to practice

### Step 2: Generate or Validate Content
- If content not provided, call skill 21-content-generate
- Validate content length ≤ 1500 characters
- Ensure content complies with medical advertising guidelines
- Add CTA if not present

### Step 3: Prepare Media
- If media_urls provided, validate each URL is accessible
- Check file size ≤ 10MB per image, ≤ 100MB per video
- Verify media format (JPEG, PNG, MP4, MOV)

### Step 4: Call GBP API
- Format request per Google Business Profile API spec
- Set languageCode: "en-IN"
- Include searchTerms for SEO
- Handle API errors:
  - 429 (quota exceeded) → queue for next day
  - 401 (auth expired) → trigger skill 03 re-auth
  - 400 (bad request) → flag for human review

### Step 5: Store Result
- Save post to database with status
- Link to content_piece if AI-generated
- Update practice metrics

### Step 6: Audit Log
- Log: practice_id, location_id, content_length, status, cost, duration
- If failed, log error message and stack trace

## Error Handling

| Error | Cause | Action |
|-------|-------|--------|
| AUTH_EXPIRED | Refresh token invalid | Trigger skill 03, notify user |
| QUOTA_EXCEEDED | Daily limit reached | Queue for next day, notify user |
| CONTENT_REJECTED | Violates GBP policy | Flag for human review |
| MEDIA_TOO_LARGE | File exceeds limit | Compress or reject |
| NETWORK_ERROR | Temporary failure | Retry with backoff |

## Implementation Notes for Coding Agents

### Database Model
```prisma
model GbpPost {
  id            String   @id @default(cuid())
  practiceId    String
  locationId    String
  gbpLocationId String
  content       String   @db.Text
  mediaUrls     String[]
  ctaType       String?
  status        String   // SCHEDULED, PUBLISHED, FAILED
  gbpPostId     String?  // Google's ID
  publishedAt   DateTime?
  failedReason  String?
  costUsd       Decimal? @db.Decimal(10, 6)
  createdAt     DateTime @default(now())
  updatedAt     DateTime @updatedAt
}

API Endpoint#

@router.post("/skills/gbp-post-create")
async def create_gbp_post(
    payload: GbpPostCreateInput,
    user: User = Depends(get_current_user)
):
    # Validate user owns practice
    # Call harness.execute_skill("04-gbp-post-create", payload)
    # Return result

Restate Workflow#

@restate.workflow
async def gbp_post_create_workflow(ctx: restate.Context, payload: dict):
    # Step 1: Validate
    practice = await ctx.run("validate", lambda: validate_practice(payload["practice_id"]))
    
    # Step 2: Generate content if needed
    if not payload.get("content"):
        content = await ctx.run("generate", lambda: ai.generate_gbp_post(practice))
    else:
        content = payload["content"]
    
    # Step 3: Publish
    result = await ctx.run("publish", lambda: gbp_api.create_post(practice, content))
    
    # Step 4: Store
    await ctx.run("store", lambda: db.gbp_post.create({...}))
    
    return result

Acceptance Criteria#

• Can create a GBP post with provided content
• Can generate AI content if none provided
• Respects 1500 character limit
• Handles auth errors gracefully
• Handles quota exceeded gracefully
• Stores audit log entry
• Updates practice metrics
• Returns within 30 seconds

### 4.2 Skill Categories

| Category | Description | Example Skills |
|----------|-------------|----------------|
| `onboarding` | Getting doctors set up | practice-setup, gbp-connect, asset-upload |
| `gbp` | Google Business Profile | post-create, post-schedule, review-reply, insights-fetch |
| `social` | Social media | connect, post-create, post-schedule |
| `citation` | Directory listings | submit, verify-nap, update |
| `site` | Landing pages | generate, update, deploy, evolve |
| `seo` | SEO operations | audit, keyword-track, competitor-analyze |
| `content` | AI content | generate, approve, publish |
| `report` | Analytics & reports | generate, email, dashboard |
| `lead` | Lead management | capture, nurture, notify |
| `billing` | Payments | subscribe, invoice, upgrade |
| `admin` | Platform operations | moderate, support, health-check |
| `platform` | System capabilities | auth, notification, dns-configure |

### 4.3 Skill Registry

The harness maintains a skill registry loaded at startup:

```python
class SkillRegistry:
    def __init__(self, skills_dir: str):
        self.skills: dict[str, Skill] = {}
        self._load_all(skills_dir)
    
    def _load_all(self, directory: str):
        for path in Path(directory).glob("**/*.md"):
            skill = self._parse_skill(path)
            self.skills[skill.id] = skill
    
    def _parse_skill(self, path: Path) -> Skill:
        content = path.read_text()
        # Parse YAML frontmatter
        # Parse markdown body
        # Extract code blocks for implementation hints
        return Skill(...)
    
    def get(self, skill_id: str) -> Skill | None:
        return self.skills.get(skill_id)
    
    def list_by_category(self, category: str) -> list[Skill]:
        return [s for s in self.skills.values() if s.category == category]
    
    def get_dependencies(self, skill_id: str) -> list[Skill]:
        skill = self.skills.get(skill_id)
        if not skill:
            return []
        return [self.skills[d] for d in skill.dependencies if d in self.skills]

4.4 Skill Execution#

class SkillHarness:
    def __init__(self, registry: SkillRegistry, restate: RestateClient):
        self.registry = registry
        self.restate = restate
        self.logger = structlog.get_logger()
    
    async def execute(self, skill_id: str, payload: dict, context: ExecutionContext) -> ExecutionResult:
        skill = self.registry.get(skill_id)
        if not skill:
            raise SkillNotFoundError(skill_id)
        
        # Check prerequisites
        for dep_id in skill.dependencies:
            dep = self.registry.get(dep_id)
            if dep and not await self._check_prerequisite(dep, payload):
                return ExecutionResult(
                    status="FAILED",
                    error=f"Prerequisite not met: {dep_id}",
                    skill_id=skill_id
                )
        
        # Log start
        self.logger.info(
            "skill_execution_started",
            skill_id=skill_id,
            practice_id=payload.get("practice_id"),
            user_id=context.user_id,
            trigger=context.trigger
        )
        
        start_time = time.time()
        
        try:
            # Dispatch to Restate workflow
            if skill.workflow_name:
                result = await self.restate.workflow(
                    skill.workflow_name,
                    payload,
                    retry_policy=skill.retry_policy
                )
            else:
                # Direct execution for simple skills
                result = await self._execute_direct(skill, payload)
            
            duration = time.time() - start_time
            
            # Log success
            self.logger.info(
                "skill_execution_completed",
                skill_id=skill_id,
                status="SUCCESS",
                duration_seconds=duration,
                cost_usd=result.get("cost_usd", 0)
            )
            
            return ExecutionResult(
                status="SUCCESS",
                data=result,
                skill_id=skill_id,
                duration_seconds=duration
            )
            
        except Exception as e:
            duration = time.time() - start_time
            
            # Log failure
            self.logger.error(
                "skill_execution_failed",
                skill_id=skill_id,
                error=str(e),
                error_type=type(e).__name__,
                duration_seconds=duration
            )
            
            # Attempt fallback if defined
            if skill.fallback_skill:
                return await self.execute(skill.fallback_skill, payload, context)
            
            return ExecutionResult(
                status="FAILED",
                error=str(e),
                skill_id=skill_id,
                duration_seconds=duration
            )

---

5. Agent Harness#

5.1 What Is the Harness?#

The harness is the runtime brain of RankFlow. It is not just a workflow engine — it is an agentic orchestrator that:

1. Understands skills — Reads skill files, knows what the platform can do
2. Plans execution — Given a goal, finds the right skills, orders them, handles dependencies
3. Dispatches work — Calls Restate workflows, external APIs, or direct functions
4. Observes everything — Logs every action, measures every metric, traces every request
5. Self-corrects — Retries, falls back, escalates, learns from failures
6. Reports back — Tells the user (and the admin) what happened and why

5.2 Harness Architecture#

# src/harness/main.py

class RankFlowHarness:
    """
    The central agent harness for RankFlow AI.
    
    Responsibilities:
    - Skill registry management
    - Execution planning and dispatch
    - Observability and audit logging
    - Self-healing and escalation
    """
    
    def __init__(self):
        self.skills = SkillRegistry("/app/skills")
        self.restate = RestateClient()
        self.audit = AuditLogger()
        self.metrics = MetricsCollector()
        self.escalation = EscalationManager()
        self.llm = LLMRouter()  # Claude → GPT-4o fallback
    
    async def handle_goal(self, goal: str, context: dict) -> HarnessResult:
        """
        High-level goal handler.
        
        Example goals:
        - "Onboard Dr. Sharma's cardiology practice"
        - "Create this week's GBP posts for all active practices"
        - "Update landing pages based on new keyword rankings"
        """
        
        # Step 1: Plan — Use LLM to break goal into skill executions
        plan = await self._plan_goal(goal, context)
        
        # Step 2: Execute — Run each step via Restate
        results = []
        for step in plan.steps:
            result = await self.execute_skill(step.skill_id, step.payload, context)
            results.append(result)
            
            if result.status == "FAILED" and not step.can_continue_on_failure:
                break
        
        # Step 3: Report — Summarize what happened
        summary = await self._generate_summary(goal, plan, results)
        
        return HarnessResult(
            goal=goal,
            plan=plan,
            results=results,
            summary=summary,
            status="SUCCESS" if all(r.status == "SUCCESS" for r in results) else "PARTIAL"
        )
    
    async def _plan_goal(self, goal: str, context: dict) -> ExecutionPlan:
        """Use LLM to plan which skills to execute."""
        
        # Load relevant skills as context
        all_skills = self.skills.list_all()
        skill_descriptions = "\n".join([
            f"- {s.id}: {s.name} (category: {s.category}, autonomous: {s.autonomous})"
            for s in all_skills
        ])
        
        prompt = f"""
You are the RankFlow AI planning engine. Given a user goal and available skills,
break the goal into a sequence of skill executions.

Available Skills:
{skill_descriptions}

User Goal: {goal}
Context: {json.dumps(context)}

Return a JSON execution plan:
{{
  "steps": [
    {{
      "skill_id": "skill-id",
      "payload": {{...}},
      "can_continue_on_failure": false,
      "reason": "why this step"
    }}
  ]
}}
"""
        
        response = await self.llm.generate(prompt, json_mode=True)
        return ExecutionPlan.parse_raw(response)

5.3 Observability in the Harness#

Every execution produces a trace:

{
  "trace_id": "abc-123-def",
  "goal": "Onboard Dr. Sharma",
  "practice_id": "practice_123",
  "user_id": "user_456",
  "started_at": "2025-01-15T09:00:00Z",
  "completed_at": "2025-01-15T09:02:30Z",
  "status": "SUCCESS",
  "steps": [
    {
      "step_id": 1,
      "skill_id": "01-practice-setup",
      "status": "SUCCESS",
      "started_at": "2025-01-15T09:00:00Z",
      "completed_at": "2025-01-15T09:00:05Z",
      "duration_ms": 5000,
      "cost_usd": 0.0,
      "inputs": {"name": "Dr. Sharma Cardiology", "city": "Mumbai"},
      "outputs": {"practice_id": "practice_123"},
      "logs": [...]
    },
    {
      "step_id": 2,
      "skill_id": "03-gbp-connect",
      "status": "WAITING",
      "started_at": "2025-01-15T09:00:05Z",
      "waiting_for": "oauth_callback",
      "auth_url": "https://accounts.google.com/..."
    }
  ],
  "total_cost_usd": 0.05,
  "total_duration_ms": 150000
}

This trace is:

• Stored in PostgreSQL (for long-term audit)
• Sent to structured logs (for real-time debugging)
• Displayed in the admin dashboard (for human oversight)
• Used by the harness for self-correction (learn from patterns)

5.4 Self-Healing#

class SelfHealingManager:
    async def handle_failure(self, trace: ExecutionTrace, step: ExecutionStep):
        """Decide what to do when a skill fails."""
        
        skill = self.skills.get(step.skill_id)
        
        # Strategy 1: Retry with modified parameters
        if step.attempt < skill.retry_policy.max_attempts:
            return HealingAction.RETRY_WITH_BACKOFF
        
        # Strategy 2: Use fallback skill
        if skill.fallback_skill:
            return HealingAction.FALLBACK(skill.fallback_skill)
        
        # Strategy 3: Use LLM to generate workaround
        workaround = await self._generate_workaround(trace, step)
        if workaround.confidence > 0.8:
            return HealingAction.WORKAROUND(workaround)
        
        # Strategy 4: Escalate to human
        return HealingAction.ESCALATE(
            reason="All automated recovery failed",
            context=trace
        )

---

6. Database Schema (Lean)#

6.1 Core Models (~15 tables)#

// prisma/schema.prisma

// ─── Auth & Users ─────────────────────────────────────────

model User {
  id            String   @id @default(cuid())
  email         String   @unique
  name          String?
  image         String?
  role          String   @default("OWNER") // OWNER, ADMIN, EDITOR, VIEWER
  createdAt     DateTime @default(now())
  updatedAt     DateTime @updatedAt
  
  practices     PracticeMember[]
  auditLogs     AuditLog[]
  
  @@map("users")
}

// ─── Practice (Tenant) ────────────────────────────────────

model Practice {
  id              String   @id @default(cuid())
  name            String
  slug            String   @unique
  type            String   @default("CLINIC") // CLINIC, HOSPITAL, DENTAL, etc.
  status          String   @default("active") // active, trial, suspended, cancelled
  ownerId         String
  
  // Subscription
  tier            String   @default("FREE") // FREE, STARTER, PRO, ENTERPRISE
  trialEndsAt     DateTime?
  subscriptionEndsAt DateTime?
  
  // Branding
  logoUrl         String?
  primaryColor    String?  @default("#2563eb")
  
  // Site
  subdomain       String   @unique @default(cuid())
  customDomain    String?  @unique
  siteTemplate    String   @default("medical-modern")
  sitePublished   Boolean  @default(false)
  
  // Settings JSON
  settings        Json     @default("{}")
  
  createdAt       DateTime @default(now())
  updatedAt       DateTime @updatedAt
  
  // Relations
  members         PracticeMember[]
  locations       Location[]
  gbpAccounts     GbpAccount[]
  socialAccounts  SocialAccount[]
  contentPieces   ContentPiece[]
  siteSections    SiteSection[]
  jobs            Job[]
  reports         Report[]
  leads           Lead[]
  auditLogs       AuditLog[]
  
  @@map("practices")
}

model PracticeMember {
  id          String   @id @default(cuid())
  practiceId  String
  userId      String
  role        String   @default("EDITOR")
  createdAt   DateTime @default(now())
  
  practice    Practice @relation(fields: [practiceId], references: [id], onDelete: Cascade)
  user        User     @relation(fields: [userId], references: [id], onDelete: Cascade)
  
  @@unique([practiceId, userId])
  @@map("practice_members")
}

// ─── Location (GBP Entity) ────────────────────────────────

model Location {
  id              String   @id @default(cuid())
  practiceId      String
  name            String
  
  // NAP
  businessName    String
  address         String
  city            String
  state           String
  postalCode      String
  country         String   @default("IN")
  phone           String
  email           String?
  website         String?
  
  // Geo
  latitude        Decimal? @db.Decimal(10, 8)
  longitude       Decimal? @db.Decimal(11, 8)
  
  // Business
  category        String   @default("Medical Clinic")
  services        String[] @default([])
  businessHours   Json     @default("{}")
  
  // SEO
  targetKeywords  String[] @default([])
  languages       String[] @default(["English", "Hindi"])
  
  isPrimary       Boolean  @default(false)
  createdAt       DateTime @default(now())
  updatedAt       DateTime @updatedAt
  
  practice        Practice @relation(fields: [practiceId], references: [id], onDelete: Cascade)
  gbpLocation     GbpLocation?
  
  @@map("locations")
}

// ─── GBP Integration ──────────────────────────────────────

model GbpAccount {
  id              String   @id @default(cuid())
  practiceId      String
  accountEmail    String
  
  // OAuth tokens (encrypted at application layer)
  accessToken     String   @db.Text
  refreshToken    String   @db.Text
  tokenExpiresAt  DateTime
  
  isActive        Boolean  @default(true)
  lastSyncedAt    DateTime?
  createdAt       DateTime @default(now())
  updatedAt       DateTime @updatedAt
  
  practice        Practice     @relation(fields: [practiceId], references: [id], onDelete: Cascade)
  locations       GbpLocation[]
  
  @@map("gbp_accounts")
}

model GbpLocation {
  id              String   @id @default(cuid())
  gbpAccountId    String
  locationId      String   // Internal location ID
  gbpLocationId   String   // Google's location ID
  name            String
  status          String   @default("ACTIVE")
  
  lastApiSyncAt   DateTime?
  createdAt       DateTime @default(now())
  updatedAt       DateTime @updatedAt
  
  gbpAccount      GbpAccount @relation(fields: [gbpAccountId], references: [id], onDelete: Cascade)
  location        Location   @relation(fields: [locationId], references: [id], onDelete: Cascade)
  posts           GbpPost[]
  reviews         Review[]
  insights        GbpInsight[]
  
  @@unique([gbpLocationId])
  @@map("gbp_locations")
}

model GbpPost {
  id              String   @id @default(cuid())
  gbpLocationId   String
  contentPieceId  String?
  
  content         String   @db.Text
  mediaUrls       String[] @default([])
  ctaType         String?
  
  status          String   @default("SCHEDULED") // SCHEDULED, PUBLISHED, FAILED
  gbpPostId       String?  // Google's post ID
  scheduledFor    DateTime?
  publishedAt     DateTime?
  failedReason    String?
  
  costUsd         Decimal? @db.Decimal(10, 6)
  createdAt       DateTime @default(now())
  updatedAt       DateTime @updatedAt
  
  gbpLocation     GbpLocation @relation(fields: [gbpLocationId], references: [id], onDelete: Cascade)
  
  @@map("gbp_posts")
}

model GbpInsight {
  id              String   @id @default(cuid())
  gbpLocationId   String
  date            DateTime @db.Date
  
  viewsSearch     Int      @default(0)
  viewsMaps       Int      @default(0)
  websiteClicks   Int      @default(0)
  phoneClicks     Int      @default(0)
  drivingDirections Int    @default(0)
  
  searchQueries   Json?    @default("[]")
  
  createdAt       DateTime @default(now())
  
  gbpLocation     GbpLocation @relation(fields: [gbpLocationId], references: [id], onDelete: Cascade)
  
  @@unique([gbpLocationId, date])
  @@map("gbp_insights")
}

// ─── Reviews ──────────────────────────────────────────────

model Review {
  id              String   @id @default(cuid())
  locationId      String
  gbpLocationId   String?
  gbpReviewId     String?  @unique
  
  reviewerName    String?
  rating          Int      @db.SmallInt
  comment         String?  @db.Text
  
  replyText       String?  @db.Text
  replyBy         String?
  replyGeneratedByAI Boolean @default(false)
  replyPublished  Boolean  @default(false)
  repliedAt       DateTime?
  
  reviewDate      DateTime
  status          String   @default("NEW") // NEW, REPLIED, FLAGGED
  
  createdAt       DateTime @default(now())
  updatedAt       DateTime @updatedAt
  
  location        Location    @relation(fields: [locationId], references: [id], onDelete: Cascade)
  gbpLocation     GbpLocation? @relation(fields: [gbpLocationId], references: [id])
  
  @@map("reviews")
}

// ─── Social Media ─────────────────────────────────────────

model SocialAccount {
  id              String   @id @default(cuid())
  practiceId      String
  platform        String   // FACEBOOK, INSTAGRAM, LINKEDIN, TWITTER
  accountName     String
  accountId       String?  // Platform's ID
  profileUrl      String?
  
  // Tokens (encrypted)
  accessToken     String   @db.Text
  refreshToken    String?  @db.Text
  tokenExpiresAt  DateTime?
  
  isActive        Boolean  @default(true)
  createdAt       DateTime @default(now())
  updatedAt       DateTime @updatedAt
  
  practice        Practice @relation(fields: [practiceId], references: [id], onDelete: Cascade)
  posts           SocialPost[]
  
  @@unique([practiceId, platform])
  @@map("social_accounts")
}

model SocialPost {
  id              String   @id @default(cuid())
  socialAccountId String
  contentPieceId  String?
  
  content         String   @db.Text
  mediaUrls       String[] @default([])
  scheduledFor    DateTime?
  publishedAt     DateTime?
  
  status          String   @default("SCHEDULED") // SCHEDULED, PUBLISHED, FAILED
  externalPostId  String?
  failedReason    String?
  
  createdAt       DateTime @default(now())
  updatedAt       DateTime @updatedAt
  
  socialAccount   SocialAccount @relation(fields: [socialAccountId], references: [id], onDelete: Cascade)
  
  @@map("social_posts")
}

// ─── Content (AI-Generated) ───────────────────────────────

model ContentPiece {
  id              String   @id @default(cuid())
  practiceId      String
  locationId      String?
  
  type            String   // GBP_POST, SOCIAL_POST, SITE_CONTENT, FAQ, SCHEMA
  status          String   @default("DRAFT") // DRAFT, PENDING_REVIEW, APPROVED, PUBLISHED
  
  title           String?
  content         String   @db.Text
  excerpt         String?  @db.Text
  
  seoTitle        String?
  seoDescription  String?
  focusKeywords   String[] @default([])
  
  aiGenerated     Boolean  @default(false)
  aiModel         String?
  aiTokensUsed    Int?
  costUsd         Decimal? @db.Decimal(10, 6)
  
  humanEdited     Boolean  @default(false)
  editedBy        String?
  editedAt        DateTime?
  
  createdAt       DateTime @default(now())
  updatedAt       DateTime @updatedAt
  
  practice        Practice @relation(fields: [practiceId], references: [id], onDelete: Cascade)
  location        Location? @relation(fields: [locationId], references: [id])
  
  @@map("content_pieces")
}

// ─── Site Sections (Landing Page) ─────────────────────────

model SiteSection {
  id              String   @id @default(cuid())
  practiceId      String
  
  sectionKey      String   // hero, about, services, testimonials, contact, faq
  sortOrder       Int      @default(0)
  content         String   @db.Text
  mediaUrls       String[] @default([])
  config          Json     @default("{}") // Section-specific settings
  isVisible       Boolean  @default(true)
  
  createdAt       DateTime @default(now())
  updatedAt       DateTime @updatedAt
  
  practice        Practice @relation(fields: [practiceId], references: [id], onDelete: Cascade)
  
  @@unique([practiceId, sectionKey])
  @@map("site_sections")
}

// ─── Jobs / Workflows ─────────────────────────────────────

model Job {
  id              String   @id @default(cuid())
  practiceId      String
  
  skillId         String   // Which skill was executed
  status          String   @default("PENDING") // PENDING, RUNNING, COMPLETED, FAILED, WAITING
  
  payload         Json     // Input data
  result          Json?    // Output data
  error           String?
  
  restateId       String?  // Restate workflow ID
  startedAt       DateTime?
  completedAt     DateTime?
  
  costUsd         Decimal? @db.Decimal(10, 6)
  createdAt       DateTime @default(now())
  updatedAt       DateTime @updatedAt
  
  practice        Practice @relation(fields: [practiceId], references: [id], onDelete: Cascade)
  
  @@index([practiceId, status])
  @@index([skillId, status])
  @@map("jobs")
}

// ─── Reports ──────────────────────────────────────────────

model Report {
  id              String   @id @default(cuid())
  practiceId      String
  
  name            String
  periodStart     DateTime
  periodEnd       DateTime
  
  status          String   @default("GENERATING") // GENERATING, READY, FAILED
  sections        Json     // Report data
  
  scoreOverall    Int?     @default(0)
  scoreGbp        Int?     @default(0)
  scoreSeo        Int?     @default(0)
  scoreReviews    Int?     @default(0)
  
  emailedAt       DateTime?
  viewedAt        DateTime?
  
  createdAt       DateTime @default(now())
  updatedAt       DateTime @updatedAt
  
  practice        Practice @relation(fields: [practiceId], references: [id], onDelete: Cascade)
  
  @@map("reports")
}

// ─── Leads ────────────────────────────────────────────────

model Lead {
  id              String   @id @default(cuid())
  practiceId      String
  
  name            String?
  phone           String?
  email           String?
  message         String?  @db.Text
  source          String   // site, gbp, social, direct
  
  status          String   @default("NEW") // NEW, CONTACTED, CONVERTED, LOST
  
  createdAt       DateTime @default(now())
  updatedAt       DateTime @updatedAt
  
  practice        Practice @relation(fields: [practiceId], references: [id], onDelete: Cascade)
  
  @@map("leads")
}

// ─── Audit Log ────────────────────────────────────────────

model AuditLog {
  id              String   @id @default(cuid())
  practiceId      String?
  userId          String?
  
  action          String   // CREATE, UPDATE, DELETE, LOGIN, SKILL_EXECUTE, etc.
  entityType      String   // table name or resource type
  entityId        String?
  
  oldValue        Json?
  newValue        Json?
  metadata        Json     @default("{}")
  
  createdAt       DateTime @default(now())
  
  practice        Practice? @relation(fields: [practiceId], references: [id], onDelete: SetNull)
  user            User?     @relation(fields: [userId], references: [id], onDelete: SetNull)
  
  @@index([practiceId, createdAt])
  @@index([userId, createdAt])
  @@map("audit_logs")
}

6.2 Schema Principles#

• No soft deletes — hard delete with audit log (simpler for agents)
• No complex enums — use String with validation (easier to modify)
• JSON for flexible config — settings, config, metadata fields
• Encrypted at app layer — tokens stored as strings, encrypted before DB
• No separate billing tables for MVP — store subscription info on Practice
• No separate AI cost tracking table — store cost on Job and ContentPiece

---

7. FastAPI Backend#

7.1 Project Structure#

rankflow-api/
├── src/
│   ├── main.py                 # FastAPI app, middleware, lifespan
│   ├── config.py               # Settings, env vars
│   ├── dependencies.py         # Auth, DB, Redis deps
│   │
│   ├── api/
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── router.py       # Aggregate all v1 routers
│   │   │   ├── auth.py         # Better Auth integration
│   │   │   ├── practices.py    # Practice CRUD
│   │   │   ├── locations.py    # Location CRUD
│   │   │   ├── gbp.py          # GBP operations
│   │   │   ├── social.py       # Social operations
│   │   │   ├── content.py      # Content management
│   │   │   ├── site.py         # Site/landing page
│   │   │   ├── skills.py       # Skill execution endpoints
│   │   │   ├── reports.py      # Report generation
│   │   │   ├── leads.py        # Lead management
│   │   │   ├── webhooks.py     # Stripe, Composio, Google
│   │   │   └── health.py       # Health checks
│   │   └── __init__.py
│   │
│   ├── harness/
│   │   ├── __init__.py
│   │   ├── main.py             # RankFlowHarness class
│   │   ├── registry.py         # SkillRegistry
│   │   ├── planner.py          # Goal → ExecutionPlan
│   │   ├── executor.py         # Skill execution engine
│   │   ├── audit.py            # Audit logging
│   │   ├── metrics.py          # Metrics collection
│   │   └── healing.py          # Self-healing logic
│   │
│   ├── workflows/
│   │   ├── __init__.py
│   │   ├── restate_client.py   # Restate connection
│   │   ├── gbp.py              # GBP workflows
│   │   ├── social.py           # Social workflows
│   │   ├── site.py             # Site generation workflows
│   │   ├── content.py          # Content generation workflows
│   │   ├── report.py           # Report workflows
│   │   └── onboarding.py       # Onboarding workflows
│   │
│   ├── services/
│   │   ├── __init__.py
│   │   ├── ai/
│   │   │   ├── __init__.py
│   │   │   ├── router.py       # Claude → GPT-4o fallback
│   │   │   ├── prompts.py      # Prompt templates
│   │   │   └── models.py       # Pydantic models
│   │   ├── gbp/
│   │   │   ├── __init__.py
│   │   │   ├── client.py       # GBP API client
│   │   │   ├── auth.py         # OAuth flow
│   │   │   └── types.py        # Pydantic models
│   │   ├── social/
│   │   │   ├── __init__.py
│   │   │   ├── composio.py     # Composio integration
│   │   │   ├── zernio.py       # Zernio scheduling
│   │   │   └── platforms.py    # Platform-specific logic
│   │   ├── seo/
│   │   │   ├── __init__.py
│   │   │   ├── dataforseo.py   # DataForSEO client
│   │   │   ├── analyzer.py     # SEO analysis
│   │   │   └── schema.py       # Schema.org generation
│   │   ├── browser/
│   │   │   ├── __init__.py
│   │   │   ├── hyperbrowser.py # Hyperbrowser API
│   │   │   ├── firecrawl.py    # Firecrawl API
│   │   │   └── scrapegraph.py  # ScrapeGraph API
│   │   ├── storage/
│   │   │   ├── __init__.py
│   │   │   └── s3.py           # S3 upload/download
│   │   └── email/
│   │       ├── __init__.py
│   │       └── resend.py       # Resend client
│   │
│   ├── db/
│   │   ├── __init__.py
│   │   ├── prisma.py           # Prisma client singleton
│   │   └── models.py            # Pydantic models for DB
│   │
│   ├── core/
│   │   ├── __init__.py
│   │   ├── auth.py             # Better Auth integration
│   │   ├── cache.py            # Redis cache wrapper
│   │   ├── crypto.py           # Encryption/decryption
│   │   ├── logging.py          # Structured logging setup
│   │   └── exceptions.py       # Custom exceptions
│   │
│   └── skills/                 # Skill definitions (loaded at runtime)
│       ├── 00-platform-overview.md
│       ├── 01-practice-setup.md
│       ├── 02-asset-upload.md
│       ├── 03-gbp-connect.md
│       ├── 04-gbp-post-create.md
│       ├── 05-gbp-post-schedule.md
│       ├── 06-gbp-review-reply.md
│       ├── 07-gbp-insights-fetch.md
│       ├── 08-social-connect.md
│       ├── 09-social-post-create.md
│       ├── 10-social-post-schedule.md
│       ├── 11-citation-submit.md
│       ├── 12-citation-verify.md
│       ├── 13-site-generate.md
│       ├── 14-site-update.md
│       ├── 15-site-evolve.md
│       ├── 16-seo-audit.md
│       ├── 17-keyword-track.md
│       ├── 18-competitor-analyze.md
│       ├── 19-report-generate.md
│       ├── 20-report-email.md
│       ├── 21-content-generate.md
│       ├── 22-content-approve.md
│       ├── 23-lead-capture.md
│       ├── 24-lead-nurture.md
│       ├── 25-dns-configure.md
│       ├── 26-billing-subscribe.md
│       ├── 27-user-invite.md
│       ├── 28-notification-send.md
│       ├── 29-error-handle.md
│       └── 30-fallback-execute.md
│
├── prisma/
│   └── schema.prisma
│
├── tests/
│   ├── unit/
│   ├── integration/
│   └── skills/                 # Skill acceptance tests
│
├── Dockerfile
├── docker-compose.yml
├── pyproject.toml
└── .env.example

7.2 FastAPI App Setup#

# src/main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from src.config import settings
from src.core.logging import setup_logging
from src.db.prisma import prisma
from src.harness.main import RankFlowHarness
from src.workflows.restate_client import RestateClient
from src.api.v1.router import api_router

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    setup_logging()
    await prisma.connect()
    app.state.harness = RankFlowHarness()
    app.state.restate = RestateClient()
    await app.state.harness.initialize()
    yield
    # Shutdown
    await prisma.disconnect()

app = FastAPI(
    title="RankFlow AI API",
    description="Agent-centric local SEO automation platform",
    version="2.0.0",
    lifespan=lifespan
)

# Middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=[settings.APP_URL],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Request ID injection
@app.middleware("http")
async def request_id_middleware(request, call_next):
    request_id = str(uuid.uuid4())
    request.state.request_id = request_id
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

# Routes
app.include_router(api_router, prefix="/api/v1")

# Health check
@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "version": "2.0.0",
        "timestamp": datetime.utcnow().isoformat()
    }

7.3 Skill Execution Endpoint#

# src/api/v1/skills.py

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel

from src.dependencies import get_current_user, get_harness
from src.harness.main import RankFlowHarness

router = APIRouter(prefix="/skills", tags=["skills"])

class ExecuteSkillRequest(BaseModel):
    skill_id: str
    payload: dict
    async_execution: bool = False  # If true, return job ID immediately

class ExecuteSkillResponse(BaseModel):
    status: str
    data: dict | None = None
    job_id: str | None = None
    error: str | None = None

@router.post("/execute", response_model=ExecuteSkillResponse)
async def execute_skill(
    request: ExecuteSkillRequest,
    user: User = Depends(get_current_user),
    harness: RankFlowHarness = Depends(get_harness)
):
    """Execute a skill by ID."""
    
    context = ExecutionContext(
        user_id=user.id,
        practice_id=request.payload.get("practice_id"),
        trigger="manual",
        request_id=getattr(request, "state", {}).get("request_id")
    )
    
    if request.async_execution:
        # Queue for background execution
        job = await harness.queue_skill(request.skill_id, request.payload, context)
        return ExecuteSkillResponse(status="QUEUED", job_id=job.id)
    
    # Synchronous execution
    result = await harness.execute(request.skill_id, request.payload, context)
    
    return ExecuteSkillResponse(
        status=result.status,
        data=result.data,
        error=result.error
    )

@router.get("/list")
async def list_skills(harness: RankFlowHarness = Depends(get_harness)):
    """List all available skills."""
    skills = harness.registry.list_all()
    return [
        {
            "id": s.id,
            "name": s.name,
            "category": s.category,
            "autonomous": s.autonomous,
            "description": s.description
        }
        for s in skills
    ]

@router.get("/{skill_id}")
async def get_skill(skill_id: str, harness: RankFlowHarness = Depends(get_harness)):
    """Get skill details."""
    skill = harness.registry.get(skill_id)
    if not skill:
        raise HTTPException(status_code=404, detail="Skill not found")
    return skill.to_dict()

---

8. Next.js Frontend#

8.1 Project Structure#

rankflow-web/
├── src/
│   ├── app/
│   │   ├── layout.tsx              # Root layout with providers
│   │   ├── page.tsx                # Marketing landing page
│   │   │
│   │   ├── (dashboard)/            # Authenticated dashboard
│   │   │   ├── layout.tsx          # Dashboard shell
│   │   │   ├── page.tsx            # Dashboard home
│   │   │   ├── practice/
│   │   │   ├── locations/
│   │   │   ├── gbp/
│   │   │   ├── social/
│   │   │   ├── content/
│   │   │   ├── site/
│   │   │   ├── reports/
│   │   │   ├── leads/
│   │   │   ├── settings/
│   │   │   └── billing/
│   │   │
│   │   ├── site/                   # Public client sites
│   │   │   └── [practiceSlug]/
│   │   │       └── page.tsx        # Dynamic site renderer
│   │   │
│   │   ├── auth/
│   │   │   ├── signin/
│   │   │   ├── signup/
│   │   │   └── callback/
│   │   │
│   │   └── api/                    # Next.js API routes (minimal)
│   │       └── webhooks/
│   │
│   ├── components/
│   │   ├── ui/                     # shadcn/ui components
│   │   ├── dashboard/              # Dashboard-specific
│   │   ├── site/                   # Site/landing page sections
│   │   └── forms/                  # Reusable forms
│   │
│   ├── lib/
│   │   ├── api.ts                  # FastAPI client (fetch wrapper)
│   │   ├── auth.ts                 # Better Auth client
│   │   ├── utils.ts                # Utilities
│   │   └── hooks.ts                # React hooks
│   │
│   └── types/
│       └── index.ts                # TypeScript types
│
├── public/
│   └── templates/                  # Site template assets
│
├── next.config.js
├── tailwind.config.ts
├── tsconfig.json
└── package.json

8.2 API Client#

// src/lib/api.ts

const API_URL = process.env.NEXT_PUBLIC_API_URL;

class RankFlowAPI {
  private async request<T>(
    endpoint: string,
    options: RequestInit = {}
  ): Promise<T> {
    const session = await authClient.getSession();
    
    const response = await fetch(`${API_URL}${endpoint}`, {
      ...options,
      headers: {
        "Content-Type": "application/json",
        ...(session?.token ? { Authorization: `Bearer ${session.token}` } : {}),
        ...options.headers,
      },
    });
    
    if (!response.ok) {
      const error = await response.json();
      throw new Error(error.message || "API request failed");
    }
    
    return response.json();
  }
  
  // Skills
  async executeSkill(skillId: string, payload: object, async = false) {
    return this.request("/api/v1/skills/execute", {
      method: "POST",
      body: JSON.stringify({ skill_id: skillId, payload, async_execution: async }),
    });
  }
  
  async listSkills() {
    return this.request("/api/v1/skills/list");
  }
  
  // Practices
  async getPractices() {
    return this.request("/api/v1/practices");
  }
  
  async getPractice(id: string) {
    return this.request(`/api/v1/practices/${id}`);
  }
  
  // GBP
  async getGbpAccounts(practiceId: string) {
    return this.request(`/api/v1/gbp/accounts?practice_id=${practiceId}`);
  }
  
  async createGbpPost(practiceId: string, data: object) {
    return this.request("/api/v1/gbp/posts", {
      method: "POST",
      body: JSON.stringify({ practice_id: practiceId, ...data }),
    });
  }
  
  // Site
  async getSiteSections(practiceId: string) {
    return this.request(`/api/v1/site/sections?practice_id=${practiceId}`);
  }
  
  async updateSiteSection(practiceId: string, sectionKey: string, data: object) {
    return this.request(`/api/v1/site/sections/${sectionKey}`, {
      method: "PUT",
      body: JSON.stringify({ practice_id: practiceId, ...data }),
    });
  }
  
  // Reports
  async getReports(practiceId: string) {
    return this.request(`/api/v1/reports?practice_id=${practiceId}`);
  }
  
  // Leads
  async getLeads(practiceId: string) {
    return this.request(`/api/v1/leads?practice_id=${practiceId}`);
  }
}

export const api = new RankFlowAPI();

8.3 Site Renderer (Dynamic Landing Pages)#

// src/app/site/[practiceSlug]/page.tsx

import { notFound } from "next/navigation";
import { Metadata } from "next";

interface Props {
  params: { practiceSlug: string };
}

export async function generateMetadata({ params }: Props): Promise<Metadata> {
  const practice = await getPracticeBySlug(params.practiceSlug);
  if (!practice) return { title: "Not Found" };
  
  return {
    title: practice.seoTitle || practice.name,
    description: practice.seoDescription,
    robots: practice.sitePublished ? "index, follow" : "noindex, nofollow",
  };
}

export default async function SitePage({ params }: Props) {
  const practice = await getPracticeBySlug(params.practiceSlug);
  if (!practice || !practice.sitePublished) {
    notFound();
  }
  
  const sections = await getSiteSections(practice.id);
  
  return (
    <div style={{ "--primary": practice.primaryColor } as React.CSSProperties}>
      <SchemaInjector practice={practice} />
      
      {sections
        .filter((s) => s.isVisible)
        .sort((a, b) => a.sortOrder - b.sortOrder)
        .map((section) => (
          <SiteSection
            key={section.sectionKey}
            section={section}
            practice={practice}
          />
        ))}
    </div>
  );
}

// Site sections map
const sectionComponents: Record<string, React.FC<any>> = {
  hero: HeroSection,
  about: AboutSection,
  services: ServicesSection,
  testimonials: TestimonialsSection,
  contact: ContactSection,
  faq: FAQSection,
  cta: CTASection,
};

function SiteSection({ section, practice }: { section: any; practice: any }) {
  const Component = sectionComponents[section.sectionKey];
  if (!Component) return null;
  
  return (
    <section id={section.sectionKey} className={`section-${section.sectionKey}`}>
      <Component content={section.content} mediaUrls={section.mediaUrls} config={section.config} practice={practice} />
    </section>
  );
}

---

9. Workflow Engine (Restate)#

9.1 Why Restate?#

Restate provides durable execution — the ability to write code that survives crashes, retries, and long waits. For an agent-centric platform, this is critical:

• No job status tables — Restate tracks state internally
• No polling — Workflows sleep and wake up on events
• No timeout anxiety — Workflows can run for days
• Transparent — You can inspect exactly where any workflow is
• Type-safe — Python SDK with full type hints

9.2 Restate Setup#

# src/workflows/restate_client.py

import restate
from restate import Context, WorkflowContext

# Restate server runs as a separate service (or embedded)
RESTATE_ENDPOINT = os.getenv("RESTATE_ENDPOINT", "http://localhost:8080")

class RestateClient:
    def __init__(self):
        self.endpoint = RESTATE_ENDPOINT
    
    async def workflow(self, name: str, payload: dict, retry_policy: dict = None):
        """Invoke a Restate workflow."""
        # Use Restate SDK to invoke
        pass
    
    async def send_event(self, workflow_id: str, event_name: str, payload: dict):
        """Send an event to a waiting workflow."""
        pass

9.3 Example Workflows#

# src/workflows/gbp.py

import restate
from restate import WorkflowContext

@restate.workflow
async def gbp_post_create_workflow(ctx: WorkflowContext, request: dict):
    """
    Durable workflow for creating a GBP post.
    Survives crashes, retries failed steps, waits for external events.
    """
    practice_id = request["practice_id"]
    location_id = request["location_id"]
    
    # Step 1: Validate practice and GBP connection (retried automatically)
    practice = await ctx.run("validate_practice", lambda: validate_practice(practice_id))
    gbp_account = await ctx.run("get_gbp_account", lambda: get_gbp_account(practice_id))
    
    if not gbp_account:
        raise restate.TerminalError("No active GBP account found")
    
    # Step 2: Generate content if not provided
    content = request.get("content")
    if not content:
        content = await ctx.run("generate_content", lambda: ai.generate_gbp_post(
            practice=practice,
            location=location_id,
            tone="professional"
        ))
    
    # Step 3: Publish to GBP (with retry)
    result = await ctx.run("publish_post", lambda: gbp_api.create_post(
        account=gbp_account,
        location_id=location_id,
        content=content,
        media_urls=request.get("media_urls", []),
        cta_type=request.get("cta_type", "BOOK")
    ))
    
    # Step 4: Store in database
    post = await ctx.run("store_post", lambda: db.gbp_post.create({
        "practice_id": practice_id,
        "gbp_location_id": location_id,
        "content": content,
        "status": "PUBLISHED",
        "gbp_post_id": result["name"],
        "published_at": datetime.utcnow(),
    }))
    
    # Step 5: Audit log
    await ctx.run("audit", lambda: audit.log({
        "action": "GBP_POST_CREATED",
        "practice_id": practice_id,
        "entity_id": post.id,
        "metadata": {"gbp_post_id": result["name"]}
    }))
    
    return {"post_id": post.id, "gbp_post_id": result["name"]}


@restate.workflow
async def gbp_oauth_workflow(ctx: WorkflowContext, request: dict):
    """
    Workflow for GBP OAuth connection.
    Can wait for days for the user to complete OAuth.
    """
    practice_id = request["practice_id"]
    
    # Step 1: Generate OAuth URL
    auth_data = await ctx.run("generate_auth_url", lambda: gbp_auth.get_auth_url(
        practice_id=practice_id,
        redirect_uri=f"{APP_URL}/api/webhooks/gbp/oauth"
    ))
    
    # Step 2: Wait for OAuth callback (can sleep for days)
    callback = await ctx.promise("oauth_callback")
    
    # Step 3: Exchange code for tokens
    tokens = await ctx.run("exchange_tokens", lambda: gbp_auth.exchange_code(
        code=callback["code"],
        state=callback["state"]
    ))
    
    # Step 4: Store tokens
    await ctx.run("store_tokens", lambda: db.gbp_account.create({
        "practice_id": practice_id,
        "account_email": tokens["email"],
        "access_token": encrypt(tokens["access_token"]),
        "refresh_token": encrypt(tokens["refresh_token"]),
        "token_expires_at": tokens["expires_at"],
    }))
    
    # Step 5: Sync locations
    locations = await ctx.run("sync_locations", lambda: gbp_api.list_locations(tokens["access_token"]))
    for loc in locations:
        await ctx.run("store_location", lambda: db.gbp_location.create({...}))
    
    return {"account_id": account.id, "locations_count": len(locations)}


@restate.workflow
async def onboarding_workflow(ctx: WorkflowContext, request: dict):
    """
    Complete practice onboarding workflow.
    Orchestrates multiple skills over time.
    """
    practice_id = request["practice_id"]
    
    # Step 1: Create practice record
    practice = await ctx.run("create_practice", lambda: db.practice.create(request))
    
    # Step 2: Generate initial site content
    await ctx.run("generate_site", lambda: site.generate(practice_id))
    
    # Step 3: Wait for GBP connection (user must complete OAuth)
    await ctx.run("notify_gbp_auth", lambda: notify_user(practice_id, "Please connect your Google Business Profile"))
    
    # Wait for GBP connection event
    await ctx.promise("gbp_connected")
    
    # Step 4: Generate first GBP post
    await ctx.run("create_first_post", lambda: harness.execute("04-gbp-post-create", {
        "practice_id": practice_id,
        "content": "Welcome to our practice! We are now accepting new patients."
    }))
    
    # Step 5: Schedule weekly posts
    await ctx.run("schedule_posts", lambda: schedule_weekly_posts(practice_id))
    
    # Step 6: Wait 1 day, then send welcome report
    await ctx.sleep(timedelta(days=1))
    await ctx.run("send_welcome_report", lambda: report.generate_welcome(practice_id))
    
    return {"practice_id": practice_id, "status": "onboarded"}

9.4 Restate Deployment#

Restate can be deployed:

1. Self-hosted (Docker container alongside the app)
2. Restate Cloud (managed, pay per invocation)

For MVP, self-hosted on the same EC2 instance:

# docker-compose.yml (Dokploy services)

services:
  restate:
    image: restatedev/restate:latest
    ports:
      - "8080:8080"
      - "9070:9070"  # Admin port
    environment:
      - RESTATE_OBSERVABILITY__LOG__FORMAT=json
    volumes:
      - restate-data:/restate-data

---

10. Landing Page Scale Plan#

10.1 Phase 1: Path-Based Routing (MVP)#

https://rankflow.ai/site/dr-sharma-cardiology-mumbai

• Single Next.js app
• Dynamic route /site/[practiceSlug]
• Content from database
• No DNS complexity
• No SSL complexity

10.2 Phase 2: Subdomain Routing#

https://dr-sharma.rankflow.ai

• Wildcard DNS: *.rankflow.ai → EC2 IP
• Next.js middleware checks Host header
• Same app, same database
• SSL via Dokploy (Let's Encrypt wildcard)

// src/middleware.ts

export function middleware(request: NextRequest) {
  const host = request.headers.get("host") || "";
  
  // Check if subdomain request
  if (host.endsWith("rankflow.ai") && host !== "rankflow.ai") {
    const subdomain = host.replace(".rankflow.ai", "");
    
    // Rewrite to site renderer
    return NextResponse.rewrite(
      new URL(`/site/${subdomain}${request.nextUrl.pathname}`, request.url)
    );
  }
  
  return NextResponse.next();
}

10.3 Phase 3: Custom Domains#

https://drsharma.com → CNAME to dr-sharma.rankflow.ai

• Doctor adds CNAME record
• Cloudflare API verifies DNS
• Dokploy provisions SSL certificate
• Next.js middleware handles custom domain

10.4 The "Living Site" Architecture#

The site is not static — it evolves based on data:

@restate.cron(every="7d")
async def evolve_site_workflow(ctx: WorkflowContext):
    """Weekly site evolution — updates content based on SEO data."""
    
    practices = await ctx.run("get_active_practices", lambda: db.practice.find_many(
        site_published=True,
        auto_update_enabled=True
    ))
    
    for practice in practices:
        # 1. Get latest SEO data
        rankings = await ctx.run("get_rankings", lambda: dataforseo.get_rankings(practice.id))
        
        # 2. Get GBP insights
        insights = await ctx.run("get_insights", lambda: gbp_api.get_insights(practice.id))
        
        # 3. AI plans updates
        update_plan = await ctx.run("plan_updates", lambda: ai.plan_site_updates(
            practice=practice,
            rankings=rankings,
            insights=insights
        ))
        
        # 4. Apply updates
        for update in update_plan.updates:
            await ctx.run(f"update_{update.section}", lambda: db.site_section.update(
                practice_id=practice.id,
                section_key=update.section,
                content=update.new_content
            ))
        
        # 5. Revalidate site
        await ctx.run("revalidate", lambda: revalidate_site(practice.subdomain))
        
        # 6. Notify doctor
        await ctx.run("notify", lambda: email.send_update_summary(
            practice_id=practice.id,
            changes=update_plan.updates
        ))

10.5 Site Templates#

Templates are React components + config:

// src/lib/site-templates.ts

export interface SiteTemplate {
  id: string;
  name: string;
  sections: string[]; // hero, about, services, testimonials, contact, faq
  defaultConfig: {
    primaryColor: string;
    fontFamily: string;
    borderRadius: string;
  };
}

export const templates: Record<string, SiteTemplate> = {
  "medical-modern": {
    id: "medical-modern",
    name: "Medical Modern",
    sections: ["hero", "about", "services", "testimonials", "faq", "contact"],
    defaultConfig: {
      primaryColor: "#2563eb",
      fontFamily: "Inter, system-ui, sans-serif",
      borderRadius: "0.5rem",
    },
  },
  "dental-clean": {
    id: "dental-clean",
    name: "Dental Clean",
    sections: ["hero", "about", "services", "gallery", "testimonials", "contact"],
    defaultConfig: {
      primaryColor: "#06b6d4",
      fontFamily: "DM Sans, system-ui, sans-serif",
      borderRadius: "9999px",
    },
  },
};

10.6 Asset Management#

Doctors upload assets via dashboard:

1. Logo → S3, optimized to WebP
2. Doctor photo → S3, face detection, optimized
3. Clinic photos → S3, gallery generation
4. Certificates → S3, displayed in about section

AI can also generate missing assets:

• No logo? → Generate simple text-based logo
• No photos? → Use GBP photos or stock images
• No doctor photo? → Prompt user or use placeholder

---

11. SEO Data Integration (DataForSEO)#

11.1 DataForSEO API#

DataForSEO provides comprehensive SEO data via REST API:

11.2 DataForSEO Client#

# src/services/seo/dataforseo.py

import httpx

class DataForSEOClient:
    def __init__(self):
        self.login = os.getenv("DATAFORSEO_LOGIN")
        self.password = os.getenv("DATAFORSEO_PASSWORD")
        self.base_url = "https://api.dataforseo.com/v3"
    
    async def _request(self, endpoint: str, data: dict) -> dict:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/{endpoint}",
                json=data,
                auth=(self.login, self.password)
            )
            response.raise_for_status()
            return response.json()
    
    async def get_serp(self, keyword: str, location: str, language: str = "en"):
        """Get SERP results for a keyword."""
        return await self._request("serp/google/organic/live/advanced", {
            "keyword": keyword,
            "location_code": location,
            "language_code": language,
            "device": "desktop",
            "os": "windows"
        })
    
    async def get_rankings(self, keywords: list[str], domain: str, location: str):
        """Check where a domain ranks for keywords."""
        results = []
        for keyword in keywords:
            serp = await self.get_serp(keyword, location)
            # Find domain in results
            position = None
            for i, result in enumerate(serp["tasks"][0]["result"][0]["items"]):
                if domain in result.get("domain", ""):
                    position = i + 1
                    break
            results.append({
                "keyword": keyword,
                "position": position,
                "date": datetime.utcnow().isoformat()
            })
        return results
    
    async def get_keyword_data(self, keyword: str, location: str):
        """Get search volume, difficulty, CPC."""
        return await self._request("keywords_data/google/search_volume/live", {
            "keywords": [keyword],
            "location_code": location,
            "language_code": "en"
        })
    
    async def get_competitors(self, domain: str, location: str):
        """Get competitor domains."""
        return await self._request("domain_analytics/competitors/live", {
            "target": domain,
            "location_code": location,
            "language_code": "en"
        })
    
    async def get_backlinks(self, domain: str):
        """Get backlink profile."""
        return await self._request("backlinks/backlinks/live", {
            "target": domain,
            "mode": "as_is"
        })

11.3 Rank Tracking Workflow#

@restate.cron(every="1d")
async def rank_tracking_workflow(ctx: WorkflowContext):
    """Daily rank tracking for all active practices."""
    
    practices = await ctx.run("get_practices", lambda: db.practice.find_many(
        status="active",
        tier={"in": ["STARTER", "PRO", "ENTERPRISE"]}
    ))
    
    for practice in practices:
        locations = await ctx.run("get_locations", lambda: db.location.find_many(
            practice_id=practice.id
        ))
        
        for location in locations:
            keywords = location.target_keywords
            if not keywords:
                continue
            
            # Get rankings
            rankings = await ctx.run("fetch_rankings", lambda: dataforseo.get_rankings(
                keywords=keywords,
                domain=practice.custom_domain or f"{practice.subdomain}.rankflow.ai",
                location=get_location_code(location.city)
            ))
            
            # Store results
            await ctx.run("store_rankings", lambda: db.rank_tracking.create_many(
                data=[{
                    "practice_id": practice.id,
                    "location_id": location.id,
                    "keyword": r["keyword"],
                    "position": r["position"],
                    "date": datetime.utcnow()
                } for r in rankings]
            ))
            
            # Check for significant changes
            for ranking in rankings:
                if ranking["position"] and ranking["position"] <= 3:
                    await ctx.run("notify_top3", lambda: notification.send(
                        practice_id=practice.id,
                        title="🎉 Top 3 Ranking!",
                        message=f"Your practice is now in the top 3 for '{ranking['keyword']}'"
                    ))

---

12. Browser Automation (API-First)#

12.1 Hyperbrowser#

Hyperbrowser provides remote browser sessions via API:

# src/services/browser/hyperbrowser.py

import httpx

class HyperbrowserClient:
    def __init__(self):
        self.api_key = os.getenv("HYPERBROWSER_API_KEY")
        self.base_url = "https://app.hyperbrowser.ai/api/v1"
    
    async def create_session(self):
        """Create a new browser session."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/sessions",
                headers={"Authorization": f"Bearer {self.api_key}"},
                json={"browser": "chrome", "headless": True}
            )
            return response.json()["session_id"]
    
    async def navigate(self, session_id: str, url: str):
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/sessions/{session_id}/navigate",
                headers={"Authorization": f"Bearer {self.api_key}"},
                json={"url": url}
            )
            return response.json()
    
    async def fill_form(self, session_id: str, selector: str, value: str):
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/sessions/{session_id}/fill",
                headers={"Authorization": f"Bearer {self.api_key}"},
                json={"selector": selector, "value": value}
            )
            return response.json()
    
    async def click(self, session_id: str, selector: str):
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/sessions/{session_id}/click",
                headers={"Authorization": f"Bearer {self.api_key}"},
                json={"selector": selector}
            )
            return response.json()
    
    async def screenshot(self, session_id: str):
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.base_url}/sessions/{session_id}/screenshot",
                headers={"Authorization": f"Bearer {self.api_key}"}
            )
            return response.content
    
    async def close_session(self, session_id: str):
        async with httpx.AsyncClient() as client:
            await client.delete(
                f"{self.base_url}/sessions/{session_id}",
                headers={"Authorization": f"Bearer {self.api_key}"}
            )

12.2 Firecrawl#

Firecrawl scrapes and extracts structured data:

# src/services/browser/firecrawl.py

import httpx

class FirecrawlClient:
    def __init__(self):
        self.api_key = os.getenv("FIRECRAWL_API_KEY")
        self.base_url = "https://api.firecrawl.dev/v1"
    
    async def scrape(self, url: str, extract_schema: dict = None):
        """Scrape a URL and optionally extract structured data."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/scrape",
                headers={"Authorization": f"Bearer {self.api_key}"},
                json={
                    "url": url,
                    "formats": ["markdown", "html"],
                    "onlyMainContent": True,
                    "extract": extract_schema
                }
            )
            return response.json()
    
    async def verify_nap(self, url: str, expected_nap: dict):
        """Verify NAP consistency on a directory listing."""
        schema = {
            "type": "object",
            "properties": {
                "business_name": {"type": "string"},
                "address": {"type": "string"},
                "phone": {"type": "string"}
            }
        }
        
        result = await self.scrape(url, extract_schema=schema)
        extracted = result["data"]["extract"]
        
        return {
            "name_match": self._fuzzy_match(expected_nap["name"], extracted.get("business_name", "")),
            "address_match": self._fuzzy_match(expected_nap["address"], extracted.get("address", "")),
            "phone_match": self._normalize_phone(expected_nap["phone"]) in self._normalize_phone(extracted.get("phone", ""))
        }

12.3 Citation Submission Workflow#

@restate.workflow
async def citation_submit_workflow(ctx: WorkflowContext, request: dict):
    """Submit practice to a directory using browser automation APIs."""
    
    practice_id = request["practice_id"]
    directory = request["directory"]  # justdial, practo, etc.
    
    practice = await ctx.run("get_practice", lambda: db.practice.find_unique(practice_id))
    location = await ctx.run("get_location", lambda: db.location.find_first(practice_id))
    
    # Use Hyperbrowser for form submission
    session_id = await ctx.run("create_session", lambda: hyperbrowser.create_session())
    
    try:
        if directory == "justdial":
            await ctx.run("navigate", lambda: hyperbrowser.navigate(session_id, "https://www.justdial.com/Free-Listing"))
            await ctx.run("fill_name", lambda: hyperbrowser.fill_form(session_id, "input[name='cnm']", location.business_name))
            await ctx.run("fill_phone", lambda: hyperbrowser.fill_form(session_id, "input[name='mn']", location.phone))
            await ctx.run("fill_city", lambda: hyperbrowser.fill_form(session_id, "input[name='city']", location.city))
            await ctx.run("submit", lambda: hyperbrowser.click(session_id, "button[type='submit']"))
            
            # Capture result
            screenshot = await ctx.run("screenshot", lambda: hyperbrowser.screenshot(session_id))
            
            # Store result
            await ctx.run("store", lambda: db.citation.create({
                "practice_id": practice_id,
                "directory": directory,
                "status": "SUBMITTED",
                "screenshot_url": await s3.upload(screenshot)
            }))
        
        elif directory == "practo":
            # Similar flow for Practo
            pass
        
    finally:
        await ctx.run("close_session", lambda: hyperbrowser.close_session(session_id))

---

13. Social Media Integration#

13.1 Composio#

Composio handles OAuth and action execution for social platforms:

# src/services/social/composio.py

import httpx

class ComposioClient:
    def __init__(self):
        self.api_key = os.getenv("COMPOSIO_API_KEY")
        self.base_url = "https://backend.composio.dev/api/v1"
    
    async def initiate_connection(self, app: str, redirect_uri: str, metadata: dict):
        """Start OAuth flow for a social platform."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/connectedAccounts",
                headers={
                    "x-api-key": self.api_key,
                    "Content-Type": "application/json"
                },
                json={
                    "appName": app,
                    "redirectUri": redirect_uri,
                    "metadata": metadata
                }
            )
            return response.json()
    
    async def execute_action(self, action: str, params: dict):
        """Execute a Composio action."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/actions/{action}/execute",
                headers={
                    "x-api-key": self.api_key,
                    "Content-Type": "application/json"
                },
                json={"input": params}
            )
            return response.json()
    
    async def publish_facebook_post(self, connection_id: str, message: str, media_urls: list = None):
        return await self.execute_action("FACEBOOK_PUBLISH_POST", {
            "connected_account_id": connection_id,
            "message": message,
            "photos": media_urls or []
        })
    
    async def publish_instagram_post(self, connection_id: str, caption: str, image_url: str):
        return await self.execute_action("INSTAGRAM_PUBLISH_POST", {
            "connected_account_id": connection_id,
            "caption": caption,
            "image_url": image_url
        })

13.2 Zernio Scheduling#

Zernio handles post scheduling:

# src/services/social/zernio.py

import httpx

class ZernioClient:
    def __init__(self):
        self.api_key = os.getenv("ZERNIO_API_KEY")
        self.base_url = "https://api.zernio.com/v1"
    
    async def schedule_posts(self, posts: list[dict]):
        """Schedule multiple posts."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/schedule/batch",
                headers={"Authorization": f"Bearer {self.api_key}"},
                json={
                    "posts": posts,
                    "webhook_url": f"{APP_URL}/api/webhooks/zernio"
                }
            )
            return response.json()

13.3 Social Post Workflow#

@restate.workflow
async def social_post_workflow(ctx: WorkflowContext, request: dict):
    """Create and publish/schedule social posts."""
    
    practice_id = request["practice_id"]
    platforms = request["platforms"]  # ["FACEBOOK", "INSTAGRAM"]
    content = request["content"]
    scheduled_for = request.get("scheduled_for")
    
    # Get connected accounts
    accounts = await ctx.run("get_accounts", lambda: db.social_account.find_many(
        practice_id=practice_id,
        platform={"in": platforms},
        is_active=True
    ))
    
    if not accounts:
        raise restate.TerminalError("No active social accounts found")
    
    posts = []
    for account in accounts:
        if scheduled_for:
            # Schedule via Zernio
            post = await ctx.run("schedule", lambda: zernio.schedule_posts([{
                "platform": account.platform,
                "account_id": account.composio_connection_id,
                "content": content,
                "scheduled_for": scheduled_for.isoformat()
            }]))
        else:
            # Publish immediately via Composio
            if account.platform == "FACEBOOK":
                post = await ctx.run("publish_fb", lambda: composio.publish_facebook_post(
                    connection_id=account.composio_connection_id,
                    message=content
                ))
            elif account.platform == "INSTAGRAM":
                post = await ctx.run("publish_ig", lambda: composio.publish_instagram_post(
                    connection_id=account.composio_connection_id,
                    caption=content,
                    image_url=request.get("image_url")
                ))
        
        posts.append({
            "platform": account.platform,
            "external_post_id": post.get("post_id"),
            "status": "SCHEDULED" if scheduled_for else "PUBLISHED"
        })
    
    # Store in database
    await ctx.run("store_posts", lambda: db.social_post.create_many(
        data=[{
            "practice_id": practice_id,
            "social_account_id": account.id,
            "content": content,
            "status": p["status"],
            "external_post_id": p["external_post_id"],
            "scheduled_for": scheduled_for
        } for account, p in zip(accounts, posts)]
    ))
    
    return {"posts": posts}

---

14. Observability & Auditability#

14.1 Structured Logging#

# src/core/logging.py

import structlog
import logging
import sys

def setup_logging():
    structlog.configure(
        processors=[
            structlog.stdlib.filter_by_level,
            structlog.stdlib.add_logger_name,
            structlog.stdlib.add_log_level,
            structlog.stdlib.PositionalArgumentsFormatter(),
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.UnicodeDecoder(),
            structlog.processors.JSONRenderer()
        ],
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )
    
    # Configure standard library logging
    logging.basicConfig(
        format="%(message)s",
        stream=sys.stdout,
        level=logging.INFO,
    )

Every log entry includes:

{
  "timestamp": "2025-01-15T09:00:00Z",
  "level": "info",
  "logger": "rankflow.harness",
  "event": "skill_execution_started",
  "skill_id": "04-gbp-post-create",
  "practice_id": "practice_123",
  "user_id": "user_456",
  "request_id": "req-789",
  "trace_id": "trace-abc",
  "duration_ms": null,
  "cost_usd": null
}

14.2 Health Checks#

# src/api/v1/health.py

from fastapi import APIRouter
from src.db.prisma import prisma
from src.core.cache import redis

router = APIRouter()

@router.get("/health")
async def health_check():
    checks = {}
    status = "healthy"
    
    # Database
    try:
        await prisma.$query_raw("SELECT 1")
        checks["database"] = "pass"
    except Exception as e:
        checks["database"] = f"fail: {str(e)}"
        status = "unhealthy"
    
    # Redis
    try:
        await redis.ping()
        checks["redis"] = "pass"
    except Exception as e:
        checks["redis"] = f"fail: {str(e)}"
        status = "unhealthy"
    
    # Restate
    try:
        # Check Restate connectivity
        checks["restate"] = "pass"
    except Exception as e:
        checks["restate"] = f"fail: {str(e)}"
        status = "degraded"
    
    # External APIs (lightweight checks)
    try:
        # Check Claude API health
        checks["claude_api"] = "pass"
    except:
        checks["claude_api"] = "fail"
        if status == "healthy":
            status = "degraded"
    
    return {
        "status": status,
        "checks": checks,
        "timestamp": datetime.utcnow().isoformat(),
        "version": "2.0.0"
    }

@router.get("/metrics")
async def metrics():
    """Prometheus-compatible metrics endpoint."""
    # Return metrics for scraping
    pass

14.3 Audit Logging#

Every significant action is logged to the AuditLog table:

# src/harness/audit.py

class AuditLogger:
    async def log(self, event: AuditEvent):
        """Log an audit event."""
        
        # Store in database
        await db.audit_log.create({
            "practice_id": event.practice_id,
            "user_id": event.user_id,
            "action": event.action,
            "entity_type": event.entity_type,
            "entity_id": event.entity_id,
            "old_value": event.old_value,
            "new_value": event.new_value,
            "metadata": event.metadata
        })
        
        # Also log to structured logger
        structlog.get_logger().info(
            "audit_event",
            practice_id=event.practice_id,
            user_id=event.user_id,
            action=event.action,
            entity_type=event.entity_type,
            entity_id=event.entity_id
        )

14.4 Admin Dashboard Observability#

The admin panel shows:

---

15. Security & Compliance#

15.1 Token Encryption#

All OAuth tokens encrypted at application layer before DB storage:

# src/core/crypto.py

from cryptography.fernet import Fernet
import os

fernet = Fernet(os.getenv("ENCRYPTION_KEY"))

def encrypt(value: str) -> str:
    return fernet.encrypt(value.encode()).decode()

def decrypt(encrypted: str) -> str:
    return fernet.decrypt(encrypted.encode()).decode()

15.2 Rate Limiting#

# src/core/ratelimit.py

from redis.asyncio import Redis

redis = Redis.from_url(os.getenv("REDIS_URL"))

async def check_rate_limit(key: str, limit: int, window: int) -> bool:
    """Simple Redis-based rate limiting."""
    current = await redis.incr(key)
    if current == 1:
        await redis.expire(key, window)
    return current <= limit

# Usage in endpoints
@app.post("/api/v1/skills/execute")
async def execute_skill(request: Request, ...):
    user_id = request.state.user.id
    if not await check_rate_limit(f"ratelimit:skills:{user_id}", 20, 60):
        raise HTTPException(429, "Rate limit exceeded")
    ...

15.3 DPDPA Compliance#

# src/core/compliance.py

async def export_user_data(practice_id: str) -> dict:
    """Export all data for a practice (DPDPA right to access)."""
    practice = await db.practice.find_unique(practice_id)
    locations = await db.location.find_many(practice_id=practice_id)
    posts = await db.gbp_post.find_many(practice_id=practice_id)
    reviews = await db.review.find_many(where={"location": {"practice_id": practice_id}})
    
    return {
        "practice": practice,
        "locations": locations,
        "posts": posts,
        "reviews": reviews,
        "exported_at": datetime.utcnow().isoformat()
    }

async def delete_user_data(practice_id: str):
    """Delete/anonymize practice data (DPDPA right to erasure)."""
    await db.practice.update(practice_id, {
        "name": "[DELETED]",
        "status": "deleted",
        "custom_domain": None,
        "subdomain": f"deleted-{datetime.utcnow().timestamp()}"
    })
    
    # Delete tokens
    await db.gbp_account.update_many(
        {"practice_id": practice_id},
        {"access_token": "[DELETED]", "refresh_token": "[DELETED]", "is_active": False}
    )
    
    await db.social_account.update_many(
        {"practice_id": practice_id},
        {"access_token": "[DELETED]", "refresh_token": "[DELETED]", "is_active": False}
    )

---

16. Project Structure#

16.1 Monorepo Layout#

rankflow/
├── apps/
│   ├── web/                    # Next.js frontend
│   │   ├── src/
│   │   ├── public/
│   │   ├── next.config.js
│   │   └── package.json
│   │
│   └── api/                    # FastAPI backend
│       ├── src/
│       ├── prisma/
│       ├── skills/
│       ├── tests/
│       ├── Dockerfile
│       └── pyproject.toml
│
├── packages/
│   ├── shared-types/           # Shared TypeScript/Python types
│   ├── ui-components/          # Shared React components
│   └── skill-schema/           # Skill file validation
│
├── infra/
│   ├── docker-compose.yml      # Local development
│   ├── dokploy-config.json     # Dokploy deployment config
│   └── aws/                    # Terraform/CDK (future)
│
├── docs/
│   ├── skills/                 # Skill definitions (source of truth)
│   ├── architecture.md
│   └── api-reference.md
│
├── scripts/
│   ├── deploy.sh
│   └── seed.sh
│
├── .github/
│   └── workflows/
│       ├── ci.yml
│       └── deploy.yml
│
├── README.md
├── Makefile
└── .env.example

---

17. Deployment Guide#

17.1 Local Development#

# 1. Clone repo
git clone https://github.com/rankflow/rankflow.git
cd rankflow

# 2. Start infrastructure
docker-compose up -d postgres redis restate

# 3. Setup API
cd apps/api
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
prisma generate
prisma migrate dev
python -m src.main

# 4. Setup Web (new terminal)
cd apps/web
npm install
npm run dev

# 5. Access
# API: http://localhost:8000
# Web: http://localhost:3000
# Restate UI: http://localhost:9070

17.2 Dokploy Deployment#

# 1. Provision EC2 instance (t3.large, Ubuntu 22.04)
# 2. Install Dokploy

curl -fsSL https://dokploy.com/install.sh | bash

# 3. Configure Dokploy via UI
# - Add GitHub integration
# - Create project "rankflow"
# - Add services:
#   - rankflow-web (Next.js)
#   - rankflow-api (FastAPI)
#   - rankflow-worker (Restate)
#   - postgres (database)
#   - redis (cache)
#   - restate (workflow engine)

# 4. Set environment variables in Dokploy UI

# 5. Deploy
# Dokploy auto-deploys on git push to main branch

17.3 CI/CD (GitHub Actions)#

# .github/workflows/deploy.yml

name: Deploy

on:
  push:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      
      - name: Setup Node
        uses: actions/setup-node@v4
        with:
          node-version: "20"
      
      - name: Test API
        run: |
          cd apps/api
          pip install -r requirements.txt
          pytest
      
      - name: Test Web
        run: |
          cd apps/web
          npm ci
          npm run test
      
      - name: Test Skills
        run: |
          cd apps/api
          python -m pytest tests/skills/

  deploy:
    needs: test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Deploy to Dokploy
        run: |
          # Trigger Dokploy deployment via webhook
          curl -X POST ${{ secrets.DOKPLOY_WEBHOOK_URL }}

---

Appendix: Skill Registry#

Core Skills (V1)#

Skill Dependency Graph#

01-practice-setup
  └── 03-gbp-connect
        └── 04-gbp-post-create
              └── 05-gbp-post-schedule
        └── 06-gbp-review-reply
        └── 07-gbp-insights-fetch
              └── 19-report-generate
                    └── 20-report-email
  └── 08-social-connect
        └── 09-social-post-create
              └── 10-social-post-schedule
  └── 13-site-generate
        └── 14-site-update
              └── 15-site-evolve
                    └── 16-seo-audit
                    └── 17-keyword-track
                    └── 18-competitor-analyze
  └── 21-content-generate
        └── 22-content-approve
              └── 04-gbp-post-create
              └── 09-social-post-create
              └── 14-site-update
  └── 23-lead-capture
        └── 24-lead-nurture

---

End of Refined Technical Specification