Saas Scaffolder

#!/usr/bin/env python3
"""Feature Flag Manager — CRUD operations on a JSON-based feature flag store.

Manage feature flags for SaaS applications: create, read, update, delete,
list, and evaluate flags with support for percentage rollouts, plan-based
targeting, and environment scoping.

Uses ONLY Python standard library. No LLM or API calls.
"""

import argparse
import json
import os
import sys
import textwrap
from copy import deepcopy
from datetime import datetime, timezone


DEFAULT_STORE = "feature_flags.json"

# ---------------------------------------------------------------------------
# Flag store operations
# ---------------------------------------------------------------------------

def load_store(path):
    """Load the flag store from a JSON file. Returns dict."""
    if not os.path.exists(path):
        return {"flags": {}, "metadata": {"created_at": now_iso(), "updated_at": now_iso()}}
    with open(path, "r") as f:
        return json.load(f)


def save_store(store, path):
    """Persist the flag store to a JSON file."""
    store["metadata"]["updated_at"] = now_iso()
    os.makedirs(os.path.dirname(path) if os.path.dirname(path) else ".", exist_ok=True)
    with open(path, "w") as f:
        json.dump(store, f, indent=2)


def now_iso():
    return datetime.now(timezone.utc).isoformat()


# ---------------------------------------------------------------------------
# CRUD helpers
# ---------------------------------------------------------------------------

def create_flag(store, key, description="", enabled=False, environments=None,
                rollout_percentage=100, allowed_plans=None, tags=None):
    """Add a new feature flag to the store."""
    if key in store["flags"]:
        return False, f"Flag '{key}' already exists. Use 'update' to modify."

    store["flags"][key] = {
        "key": key,
        "description": description,
        "enabled": enabled,
        "environments": environments or ["development", "staging", "production"],
        "rollout_percentage": max(0, min(100, rollout_percentage)),
        "allowed_plans": allowed_plans or ["free", "pro", "enterprise"],
        "tags": tags or [],
        "created_at": now_iso(),
        "updated_at": now_iso(),
    }
    return True, f"Flag '{key}' created."


def get_flag(store, key):
    """Retrieve a single flag by key."""
    flag = store["flags"].get(key)
    if not flag:
        return None, f"Flag '{key}' not found."
    return deepcopy(flag), None


def update_flag(store, key, **kwargs):
    """Update fields of an existing flag."""
    if key not in store["flags"]:
        return False, f"Flag '{key}' not found."

    flag = store["flags"][key]
    updatable = ("description", "enabled", "environments", "rollout_percentage", "allowed_plans", "tags")
    changed = []
    for field in updatable:
        if field in kwargs and kwargs[field] is not None:
            old_val = flag.get(field)
            new_val = kwargs[field]
            if field == "rollout_percentage":
                new_val = max(0, min(100, new_val))
            if old_val != new_val:
                flag[field] = new_val
                changed.append(field)

    if not changed:
        return True, f"Flag '{key}' unchanged (no new values)."

    flag["updated_at"] = now_iso()
    return True, f"Flag '{key}' updated: {', '.join(changed)}."


def delete_flag(store, key):
    """Remove a flag from the store."""
    if key not in store["flags"]:
        return False, f"Flag '{key}' not found."
    del store["flags"][key]
    return True, f"Flag '{key}' deleted."


def list_flags(store, tag=None, environment=None, enabled_only=False):
    """List flags with optional filters."""
    results = []
    for flag in store["flags"].values():
        if tag and tag not in flag.get("tags", []):
            continue
        if environment and environment not in flag.get("environments", []):
            continue
        if enabled_only and not flag.get("enabled", False):
            continue
        results.append(deepcopy(flag))
    results.sort(key=lambda f: f["key"])
    return results


def evaluate_flag(store, key, environment="production", plan="free", user_hash=None):
    """Evaluate whether a flag is active for given context."""
    flag = store["flags"].get(key)
    if not flag:
        return {"active": False, "reason": "flag_not_found"}

    if not flag.get("enabled", False):
        return {"active": False, "reason": "flag_disabled"}

    if environment not in flag.get("environments", []):
        return {"active": False, "reason": f"environment '{environment}' not targeted"}

    if plan not in flag.get("allowed_plans", []):
        return {"active": False, "reason": f"plan '{plan}' not allowed"}

    pct = flag.get("rollout_percentage", 100)
    if pct < 100:
        if user_hash is not None:
            bucket = hash(f"{key}:{user_hash}") % 100
            if bucket >= pct:
                return {"active": False, "reason": f"user outside {pct}% rollout"}
        else:
            return {"active": True, "reason": f"rollout {pct}% (no user_hash to evaluate)"}

    return {"active": True, "reason": "all_checks_passed"}


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description="Manage feature flag configurations for SaaS projects (CRUD on a JSON store).",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=textwrap.dedent("""\
            examples:
              %(prog)s create --key dark-mode --description "Dark mode toggle" --enabled
              %(prog)s list --enabled-only
              %(prog)s update --key dark-mode --rollout-percentage 50
              %(prog)s evaluate --key dark-mode --environment production --plan pro
              %(prog)s delete --key dark-mode
        """),
    )

    parser.add_argument("--store", default=DEFAULT_STORE, help=f"Path to flag store JSON (default: {DEFAULT_STORE})")
    parser.add_argument("--json", action="store_true", dest="json_output", help="Output results as JSON")

    sub = parser.add_subparsers(dest="command", help="Operation to perform")

    # -- create --
    p_create = sub.add_parser("create", help="Create a new feature flag")
    p_create.add_argument("--key", required=True, help="Unique flag key (e.g. dark-mode)")
    p_create.add_argument("--description", default="", help="Human-readable description")
    p_create.add_argument("--enabled", action="store_true", help="Enable the flag immediately")
    p_create.add_argument("--environments", nargs="*", default=None,
                          help="Target environments (default: development staging production)")
    p_create.add_argument("--rollout-percentage", type=int, default=100, help="Rollout percentage 0-100")
    p_create.add_argument("--allowed-plans", nargs="*", default=None,
                          help="Plans that can see this flag (default: free pro enterprise)")
    p_create.add_argument("--tags", nargs="*", default=None, help="Tags for filtering")

    # -- get --
    p_get = sub.add_parser("get", help="Get a single flag by key")
    p_get.add_argument("--key", required=True, help="Flag key to retrieve")

    # -- update --
    p_update = sub.add_parser("update", help="Update an existing flag")
    p_update.add_argument("--key", required=True, help="Flag key to update")
    p_update.add_argument("--description", default=None)
    p_update.add_argument("--enabled", default=None, type=lambda v: v.lower() in ("true", "1", "yes"),
                          help="true/false")
    p_update.add_argument("--environments", nargs="*", default=None)
    p_update.add_argument("--rollout-percentage", type=int, default=None)
    p_update.add_argument("--allowed-plans", nargs="*", default=None)
    p_update.add_argument("--tags", nargs="*", default=None)

    # -- delete --
    p_delete = sub.add_parser("delete", help="Delete a feature flag")
    p_delete.add_argument("--key", required=True, help="Flag key to delete")

    # -- list --
    p_list = sub.add_parser("list", help="List feature flags")
    p_list.add_argument("--tag", default=None, help="Filter by tag")
    p_list.add_argument("--environment", default=None, help="Filter by target environment")
    p_list.add_argument("--enabled-only", action="store_true", help="Show only enabled flags")

    # -- evaluate --
    p_eval = sub.add_parser("evaluate", help="Evaluate a flag for a given context")
    p_eval.add_argument("--key", required=True, help="Flag key to evaluate")
    p_eval.add_argument("--environment", default="production", help="Environment context")
    p_eval.add_argument("--plan", default="free", help="Tenant plan context")
    p_eval.add_argument("--user-hash", default=None, help="User identifier for rollout bucketing")

    args = parser.parse_args()

    if not args.command:
        parser.print_help()
        sys.exit(1)

    store = load_store(args.store)
    output = {}

    if args.command == "create":
        ok, msg = create_flag(
            store, args.key, description=args.description, enabled=args.enabled,
            environments=args.environments, rollout_percentage=args.rollout_percentage,
            allowed_plans=args.allowed_plans, tags=args.tags,
        )
        save_store(store, args.store)
        output = {"success": ok, "message": msg, "flag": store["flags"].get(args.key)}

    elif args.command == "get":
        flag, err = get_flag(store, args.key)
        output = {"success": flag is not None, "flag": flag, "error": err}

    elif args.command == "update":
        ok, msg = update_flag(
            store, args.key, description=args.description, enabled=args.enabled,
            environments=args.environments, rollout_percentage=args.rollout_percentage,
            allowed_plans=args.allowed_plans, tags=args.tags,
        )
        save_store(store, args.store)
        output = {"success": ok, "message": msg, "flag": store["flags"].get(args.key)}

    elif args.command == "delete":
        ok, msg = delete_flag(store, args.key)
        save_store(store, args.store)
        output = {"success": ok, "message": msg}

    elif args.command == "list":
        flags = list_flags(store, tag=args.tag, environment=args.environment,
                           enabled_only=args.enabled_only)
        output = {"success": True, "count": len(flags), "flags": flags}

    elif args.command == "evaluate":
        result = evaluate_flag(store, args.key, environment=args.environment,
                               plan=args.plan, user_hash=args.user_hash)
        output = {"success": True, "key": args.key, "evaluation": result}

    # --- Output ---
    if args.json_output:
        print(json.dumps(output, indent=2))
    else:
        _print_human(args.command, output)


def _print_human(command, output):
    """Pretty-print results for human consumption."""
    if not output.get("success", False) and output.get("error"):
        print(f"Error: {output['error']}")
        sys.exit(1)

    if command in ("create", "update", "delete"):
        print(output.get("message", "Done."))
        flag = output.get("flag")
        if flag:
            _print_flag_summary(flag)

    elif command == "get":
        flag = output.get("flag")
        if flag:
            _print_flag_summary(flag)
        else:
            print(f"Error: {output.get('error')}")
            sys.exit(1)

    elif command == "list":
        flags = output.get("flags", [])
        print(f"Feature Flags ({output.get('count', 0)} total)")
        print("=" * 70)
        if not flags:
            print("  (none)")
        for flag in flags:
            status = "ON " if flag["enabled"] else "OFF"
            pct = flag.get("rollout_percentage", 100)
            plans = ", ".join(flag.get("allowed_plans", []))
            envs = ", ".join(flag.get("environments", []))
            print(f"  [{status}] {flag['key']:<30} rollout={pct:>3}%  plans=[{plans}]")
            if flag.get("description"):
                print(f"        {flag['description']}")

    elif command == "evaluate":
        ev = output.get("evaluation", {})
        key = output.get("key", "?")
        active = ev.get("active", False)
        reason = ev.get("reason", "")
        symbol = "ACTIVE" if active else "INACTIVE"
        print(f"Flag '{key}': {symbol}")
        print(f"  Reason: {reason}")


def _print_flag_summary(flag):
    """Print a single flag in a readable format."""
    status = "ENABLED" if flag["enabled"] else "DISABLED"
    print(f"\n  Key:         {flag['key']}")
    print(f"  Status:      {status}")
    print(f"  Description: {flag.get('description', '(none)')}")
    print(f"  Rollout:     {flag.get('rollout_percentage', 100)}%")
    print(f"  Plans:       {', '.join(flag.get('allowed_plans', []))}")
    print(f"  Envs:        {', '.join(flag.get('environments', []))}")
    print(f"  Tags:        {', '.join(flag.get('tags', [])) or '(none)'}")
    print(f"  Updated:     {flag.get('updated_at', 'N/A')}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""SaaS Project Scaffolder — Generate production-ready SaaS project structure.

Creates a complete SaaS project directory tree with authentication, billing,
multi-tenancy, and dashboard boilerplate. Outputs the file tree and generates
key configuration/boilerplate files on disk.

Uses ONLY Python standard library. No LLM or API calls.
"""

import argparse
import json
import os
import sys
import textwrap
from datetime import datetime, timezone


# ---------------------------------------------------------------------------
# Templates for generated files
# ---------------------------------------------------------------------------

ENV_EXAMPLE = """\
# ─── App ───
NEXT_PUBLIC_APP_URL=http://localhost:3000
NEXTAUTH_SECRET=           # openssl rand -base64 32
NEXTAUTH_URL=http://localhost:3000

# ─── Database ───
DATABASE_URL=              # postgresql://user:pass@host/db?sslmode=require

# ─── OAuth Providers ───
GOOGLE_CLIENT_ID=
GOOGLE_CLIENT_SECRET=
GITHUB_CLIENT_ID=
GITHUB_CLIENT_SECRET=

# ─── Stripe ───
STRIPE_SECRET_KEY=sk_test_...
STRIPE_WEBHOOK_SECRET=whsec_...
NEXT_PUBLIC_STRIPE_PUBLISHABLE_KEY=pk_test_...
STRIPE_PRO_MONTHLY_PRICE_ID=price_...
STRIPE_PRO_YEARLY_PRICE_ID=price_...

# ─── Email ───
RESEND_API_KEY=re_...
"""

PACKAGE_JSON_TMPL = """\
{{
  "name": "{name}",
  "version": "0.1.0",
  "private": true,
  "scripts": {{
    "dev": "next dev",
    "build": "next build",
    "start": "next start",
    "lint": "next lint",
    "db:push": "drizzle-kit push",
    "db:generate": "drizzle-kit generate",
    "db:seed": "tsx db/seed.ts"
  }},
  "dependencies": {{
    "next": "^15.0.0",
    "react": "^19.0.0",
    "react-dom": "^19.0.0",
    "@auth/drizzle-adapter": "^1.0.0",
    "next-auth": "^5.0.0",
    "drizzle-orm": "^0.35.0",
    "stripe": "^17.0.0",
    "zod": "^3.23.0",
    "@paralleldrive/cuid2": "^2.2.0"
  }},
  "devDependencies": {{
    "typescript": "^5.6.0",
    "drizzle-kit": "^0.27.0",
    "tailwindcss": "^3.4.0",
    "postcss": "^8.4.0",
    "autoprefixer": "^10.4.0",
    "@types/node": "^22.0.0",
    "@types/react": "^19.0.0",
    "tsx": "^4.0.0",
    "eslint": "^9.0.0",
    "prettier": "^3.3.0"
  }}
}}
"""

MIDDLEWARE_TS = """\
import { auth } from '@/lib/auth'
import { NextResponse } from 'next/server'

export default auth((req) => {
  const { pathname } = req.nextUrl
  const isAuthenticated = !!req.auth

  if (pathname.startsWith('/dashboard') || pathname.startsWith('/settings')) {
    if (!isAuthenticated) {
      return NextResponse.redirect(new URL('/login', req.url))
    }
  }

  if ((pathname === '/login' || pathname === '/register') && isAuthenticated) {
    return NextResponse.redirect(new URL('/dashboard', req.url))
  }

  return NextResponse.next()
})

export const config = {
  matcher: ['/dashboard/:path*', '/settings/:path*', '/login', '/register'],
}
"""

DB_SCHEMA_TS = """\
import { pgTable, text, timestamp, integer, boolean, uniqueIndex, index } from 'drizzle-orm/pg-core'
import { createId } from '@paralleldrive/cuid2'

export const workspaces = pgTable('workspaces', {
  id: text('id').primaryKey().$defaultFn(createId),
  name: text('name').notNull(),
  slug: text('slug').notNull(),
  plan: text('plan').notNull().default('free'),
  stripeCustomerId: text('stripe_customer_id').unique(),
  stripeSubscriptionId: text('stripe_subscription_id'),
  stripePriceId: text('stripe_price_id'),
  stripeCurrentPeriodEnd: timestamp('stripe_current_period_end'),
  createdAt: timestamp('created_at', { withTimezone: true }).defaultNow().notNull(),
  updatedAt: timestamp('updated_at', { withTimezone: true }).defaultNow().notNull(),
}, (t) => [
  uniqueIndex('workspaces_slug_idx').on(t.slug),
])

export const users = pgTable('users', {
  id: text('id').primaryKey().$defaultFn(createId),
  email: text('email').notNull().unique(),
  name: text('name'),
  avatarUrl: text('avatar_url'),
  emailVerified: timestamp('email_verified', { withTimezone: true }),
  createdAt: timestamp('created_at', { withTimezone: true }).defaultNow().notNull(),
})

export const workspaceMembers = pgTable('workspace_members', {
  id: text('id').primaryKey().$defaultFn(createId),
  workspaceId: text('workspace_id').notNull().references(() => workspaces.id, { onDelete: 'cascade' }),
  userId: text('user_id').notNull().references(() => users.id, { onDelete: 'cascade' }),
  role: text('role').notNull().default('member'),
  joinedAt: timestamp('joined_at', { withTimezone: true }).defaultNow().notNull(),
}, (t) => [
  uniqueIndex('workspace_members_unique').on(t.workspaceId, t.userId),
  index('workspace_members_workspace_idx').on(t.workspaceId),
])
"""


# ---------------------------------------------------------------------------
# Directory tree definition
# ---------------------------------------------------------------------------

def build_tree(name, auth_provider, db_provider, payment_provider, tenancy, features):
    """Return a nested dict representing the SaaS project tree."""
    tree = {
        "app": {
            "(auth)": {
                "login": {"page.tsx": None},
                "register": {"page.tsx": None},
                "forgot-password": {"page.tsx": None},
                "layout.tsx": None,
            },
            "(dashboard)": {
                "dashboard": {"page.tsx": None},
                "settings": {
                    "page.tsx": None,
                    "billing": {"page.tsx": None},
                    "team": {"page.tsx": None},
                },
                "layout.tsx": None,
            },
            "(marketing)": {
                "page.tsx": None,
                "pricing": {"page.tsx": None},
                "layout.tsx": None,
            },
            "api": {
                "auth": {"[...nextauth]": {"route.ts": None}},
                "webhooks": {"stripe": {"route.ts": None}},
                "billing": {
                    "checkout": {"route.ts": None},
                    "portal": {"route.ts": None},
                },
                "health": {"route.ts": None},
            },
            "layout.tsx": None,
            "not-found.tsx": None,
        },
        "components": {
            "ui": {},
            "auth": {"login-form.tsx": None, "register-form.tsx": None},
            "dashboard": {"sidebar.tsx": None, "header.tsx": None, "stats-card.tsx": None},
            "marketing": {"hero.tsx": None, "features.tsx": None, "pricing-card.tsx": None, "footer.tsx": None},
            "billing": {"plan-card.tsx": None, "usage-meter.tsx": None},
        },
        "lib": {
            "auth.ts": None,
            "db.ts": None,
            "stripe.ts": None,
            "validations.ts": None,
            "utils.ts": None,
        },
        "db": {
            "schema.ts": None,
            "migrations": {},
            "seed.ts": None,
        },
        "hooks": {"use-subscription.ts": None, "use-current-user.ts": None},
        "types": {"index.ts": None},
        "middleware.ts": None,
        ".env.example": None,
        "drizzle.config.ts": None,
        "tailwind.config.ts": None,
        "next.config.ts": None,
        "package.json": None,
        "tsconfig.json": None,
        "README.md": None,
    }

    # Add feature-specific directories
    for feat in features:
        slug = feat.strip().lower().replace(" ", "-")
        if slug:
            tree["app"]["(dashboard)"][slug] = {"page.tsx": None}

    if tenancy == "none":
        tree["app"]["(dashboard)"]["settings"].pop("team", None)

    if payment_provider == "none":
        tree["app"]["api"].pop("billing", None)
        tree["app"]["api"]["webhooks"].pop("stripe", None)
        tree["app"]["(dashboard)"]["settings"].pop("billing", None)
        tree["components"].pop("billing", None)
        tree["lib"].pop("stripe.ts", None)
        tree["hooks"].pop("use-subscription.ts", None)

    return tree


def render_tree(node, prefix="", is_last=True, name=""):
    """Render a nested dict as an ASCII tree string."""
    lines = []
    if name:
        connector = "└── " if is_last else "├── "
        lines.append(f"{prefix}{connector}{name}{'/' if isinstance(node, dict) else ''}")
        prefix += "    " if is_last else "│   "

    if isinstance(node, dict):
        entries = sorted(node.keys())
        for i, key in enumerate(entries):
            last = i == len(entries) - 1
            lines.extend(render_tree(node[key], prefix, last, key))
    return lines


# ---------------------------------------------------------------------------
# File generation on disk
# ---------------------------------------------------------------------------

FILE_MAP = {
    ".env.example": ENV_EXAMPLE,
    "middleware.ts": MIDDLEWARE_TS,
    "db/schema.ts": DB_SCHEMA_TS,
}


def write_project(output_dir, tree, project_name):
    """Write the project skeleton to disk."""
    files_written = []

    def _walk(node, current_path):
        if isinstance(node, dict):
            os.makedirs(current_path, exist_ok=True)
            for child_name, child_node in node.items():
                _walk(child_node, os.path.join(current_path, child_name))
        else:
            rel = os.path.relpath(current_path, output_dir)
            content = FILE_MAP.get(rel, "")
            if rel == "package.json":
                content = PACKAGE_JSON_TMPL.format(name=project_name)
            os.makedirs(os.path.dirname(current_path), exist_ok=True)
            with open(current_path, "w") as f:
                f.write(content)
            files_written.append(rel)

    _walk(tree, output_dir)
    return files_written


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description="Generate a production-ready SaaS project structure with auth, billing, and multi-tenancy.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=textwrap.dedent("""\
            examples:
              %(prog)s --name my-saas --output ./my-saas
              %(prog)s --name acme --auth clerk --db supabase --payments none --json
              %(prog)s --name startup --features "analytics,reports,integrations"
        """),
    )

    parser.add_argument("--name", required=True, help="Project name (used for directory and package.json)")
    parser.add_argument("--output", default=None, help="Output directory (default: ./<name>)")
    parser.add_argument("--auth", choices=["nextauth", "clerk", "supabase"], default="nextauth",
                        help="Authentication provider (default: nextauth)")
    parser.add_argument("--db", choices=["neondb", "supabase", "planetscale", "turso"], default="neondb",
                        help="Database provider (default: neondb)")
    parser.add_argument("--payments", choices=["stripe", "lemonsqueezy", "none"], default="stripe",
                        help="Payment provider (default: stripe)")
    parser.add_argument("--tenancy", choices=["workspace", "organization", "none"], default="workspace",
                        help="Multi-tenancy model (default: workspace)")
    parser.add_argument("--features", default="", help="Comma-separated list of additional feature pages")
    parser.add_argument("--dry-run", action="store_true", help="Show tree without writing files to disk")
    parser.add_argument("--json", action="store_true", dest="json_output", help="Output results as JSON")

    args = parser.parse_args()

    output_dir = args.output or os.path.join(".", args.name)
    features = [f.strip() for f in args.features.split(",") if f.strip()]

    tree = build_tree(args.name, args.auth, args.db, args.payments, args.tenancy, features)

    tree_lines = render_tree(tree, name=args.name)
    tree_str = "\n".join(tree_lines)

    files_written = []
    if not args.dry_run:
        files_written = write_project(output_dir, tree, args.name)

    # Build result
    result = {
        "project_name": args.name,
        "output_directory": os.path.abspath(output_dir),
        "auth_provider": args.auth,
        "database_provider": args.db,
        "payment_provider": args.payments,
        "tenancy_model": args.tenancy,
        "extra_features": features,
        "dry_run": args.dry_run,
        "files_written": len(files_written),
        "file_list": files_written,
        "tree": tree_str,
        "generated_at": datetime.now(timezone.utc).isoformat(),
    }

    if args.json_output:
        print(json.dumps(result, indent=2))
    else:
        print(f"SaaS Scaffolder — Project: {args.name}")
        print("=" * 60)
        print(f"  Auth:       {args.auth}")
        print(f"  Database:   {args.db}")
        print(f"  Payments:   {args.payments}")
        print(f"  Tenancy:    {args.tenancy}")
        if features:
            print(f"  Features:   {', '.join(features)}")
        print(f"  Output:     {os.path.abspath(output_dir)}")
        print(f"  Dry run:    {args.dry_run}")
        print()
        print("Project Tree:")
        print("-" * 60)
        print(tree_str)
        print("-" * 60)
        if not args.dry_run:
            print(f"\n{len(files_written)} files written to {os.path.abspath(output_dir)}")
            key_files = [f for f in files_written if f in (".env.example", "middleware.ts", "db/schema.ts", "package.json")]
            if key_files:
                print("\nKey files with boilerplate content:")
                for kf in key_files:
                    print(f"  - {kf}")
        else:
            print("\n(dry-run mode — no files written)")
        print(f"\nGenerated at {result['generated_at']}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Tenant Configuration Validator — Detect multi-tenant isolation issues.

Validates multi-tenant configuration files and source code for common
isolation problems: missing tenant scoping on queries, cross-tenant data
leaks, unscoped API routes, missing workspace context, and configuration
drift between tenants.

Uses ONLY Python standard library. No LLM or API calls.
"""

import argparse
import json
import os
import re
import sys
import textwrap
from datetime import datetime, timezone


# ---------------------------------------------------------------------------
# Validation rules
# ---------------------------------------------------------------------------

SEVERITY_CRITICAL = "critical"
SEVERITY_WARNING = "warning"
SEVERITY_INFO = "info"

SEVERITY_ORDER = {SEVERITY_CRITICAL: 0, SEVERITY_WARNING: 1, SEVERITY_INFO: 2}


def validate_config_file(config_path):
    """Validate a tenant configuration JSON file for structural issues."""
    findings = []

    if not os.path.exists(config_path):
        findings.append(finding("config_missing", SEVERITY_CRITICAL,
                                f"Configuration file not found: {config_path}",
                                fix="Create tenant configuration at the expected path."))
        return findings

    try:
        with open(config_path, "r") as f:
            config = json.load(f)
    except json.JSONDecodeError as e:
        findings.append(finding("config_invalid_json", SEVERITY_CRITICAL,
                                f"Invalid JSON in {config_path}: {e}",
                                fix="Fix JSON syntax errors in the configuration file."))
        return findings

    # Check for required top-level keys
    required_keys = ["tenants", "isolation_mode", "default_plan"]
    for key in required_keys:
        if key not in config:
            findings.append(finding("config_missing_key", SEVERITY_WARNING,
                                    f"Missing required key '{key}' in configuration.",
                                    fix=f"Add '{key}' to the tenant configuration root."))

    # Validate tenants array
    tenants = config.get("tenants", [])
    if not isinstance(tenants, list):
        findings.append(finding("tenants_not_array", SEVERITY_CRITICAL,
                                "'tenants' must be an array.",
                                fix="Change 'tenants' value to a JSON array."))
        return findings

    if not tenants:
        findings.append(finding("tenants_empty", SEVERITY_WARNING,
                                "No tenants defined in configuration.",
                                fix="Add at least one tenant to the 'tenants' array."))

    # Validate each tenant
    seen_ids = set()
    seen_slugs = set()
    seen_domains = set()

    tenant_required = ["id", "name", "slug", "plan"]

    for i, tenant in enumerate(tenants):
        prefix = f"tenants[{i}]"

        if not isinstance(tenant, dict):
            findings.append(finding("tenant_not_object", SEVERITY_CRITICAL,
                                    f"{prefix}: Tenant entry must be an object.",
                                    fix=f"Ensure {prefix} is a JSON object with id, name, slug, plan."))
            continue

        # Required fields
        for key in tenant_required:
            if key not in tenant:
                findings.append(finding("tenant_missing_field", SEVERITY_WARNING,
                                        f"{prefix}: Missing required field '{key}'.",
                                        fix=f"Add '{key}' to {prefix}."))

        # Unique ID check
        tid = tenant.get("id")
        if tid:
            if tid in seen_ids:
                findings.append(finding("tenant_duplicate_id", SEVERITY_CRITICAL,
                                        f"{prefix}: Duplicate tenant ID '{tid}'.",
                                        fix="Ensure every tenant has a unique 'id'."))
            seen_ids.add(tid)

        # Unique slug check
        slug = tenant.get("slug")
        if slug:
            if slug in seen_slugs:
                findings.append(finding("tenant_duplicate_slug", SEVERITY_CRITICAL,
                                        f"{prefix}: Duplicate tenant slug '{slug}'.",
                                        fix="Ensure every tenant has a unique 'slug'."))
            seen_slugs.add(slug)

        # Custom domain uniqueness
        domain = tenant.get("custom_domain")
        if domain:
            if domain in seen_domains:
                findings.append(finding("tenant_duplicate_domain", SEVERITY_CRITICAL,
                                        f"{prefix}: Duplicate custom domain '{domain}'.",
                                        fix="Each custom domain must map to exactly one tenant."))
            seen_domains.add(domain)

        # Plan validation
        plan = tenant.get("plan", "")
        valid_plans = config.get("valid_plans", ["free", "pro", "enterprise"])
        if plan and plan not in valid_plans:
            findings.append(finding("tenant_invalid_plan", SEVERITY_WARNING,
                                    f"{prefix}: Plan '{plan}' not in valid plans {valid_plans}.",
                                    fix=f"Set plan to one of: {', '.join(valid_plans)}."))

        # Feature flags reference check
        flags = tenant.get("feature_flags", {})
        if isinstance(flags, dict):
            global_flags = config.get("feature_flags", {})
            for flag_key in flags:
                if global_flags and flag_key not in global_flags:
                    findings.append(finding("tenant_unknown_flag", SEVERITY_INFO,
                                            f"{prefix}: Feature flag '{flag_key}' not in global registry.",
                                            fix="Add the flag to the global 'feature_flags' section or remove from tenant."))

        # Database isolation check
        db_config = tenant.get("database", {})
        if isinstance(db_config, dict):
            isolation = config.get("isolation_mode", "shared")
            if isolation == "dedicated" and not db_config.get("connection_string"):
                findings.append(finding("tenant_missing_db", SEVERITY_CRITICAL,
                                        f"{prefix}: Dedicated isolation requires a 'connection_string' in database config.",
                                        fix="Add 'database.connection_string' for dedicated isolation mode."))
            if isolation == "schema" and not db_config.get("schema_name"):
                findings.append(finding("tenant_missing_schema", SEVERITY_WARNING,
                                        f"{prefix}: Schema isolation requires 'schema_name' in database config.",
                                        fix="Add 'database.schema_name' for schema-based isolation."))

    # Cross-tenant checks
    isolation = config.get("isolation_mode", "shared")
    if isolation not in ("shared", "schema", "dedicated"):
        findings.append(finding("invalid_isolation_mode", SEVERITY_WARNING,
                                f"Unknown isolation_mode '{isolation}'. Expected: shared, schema, dedicated.",
                                fix="Set isolation_mode to one of: shared, schema, dedicated."))

    return findings


def scan_source_for_issues(src_dir, extensions=None):
    """Scan source files for common tenant-isolation anti-patterns."""
    findings = []

    if not os.path.isdir(src_dir):
        findings.append(finding("src_dir_missing", SEVERITY_WARNING,
                                f"Source directory not found: {src_dir}",
                                fix="Provide a valid source directory with --src."))
        return findings

    if extensions is None:
        extensions = (".ts", ".tsx", ".js", ".jsx")

    # Patterns to detect
    patterns = [
        {
            "id": "unscoped_query",
            "regex": re.compile(r"db\.(select|query|delete|update)\b(?!.*workspaceId)(?!.*tenantId)(?!.*workspace_id)(?!.*tenant_id)", re.IGNORECASE),
            "severity": SEVERITY_CRITICAL,
            "message": "Database query without tenant scoping (missing workspaceId/tenantId filter).",
            "fix": "Add a WHERE clause filtering by workspaceId or tenantId.",
        },
        {
            "id": "hardcoded_tenant",
            "regex": re.compile(r"""(['"])tenant[_-]?id\1\s*[:=]\s*(['"])[a-zA-Z0-9_-]+\2"""),
            "severity": SEVERITY_WARNING,
            "message": "Hardcoded tenant/workspace ID detected.",
            "fix": "Replace hardcoded ID with dynamic tenant resolution from session or context.",
        },
        {
            "id": "missing_auth_check",
            "regex": re.compile(r"export\s+(async\s+)?function\s+(GET|POST|PUT|PATCH|DELETE)\b(?!.*auth\(\))"),
            "severity": SEVERITY_WARNING,
            "message": "API route handler without auth() check.",
            "fix": "Add `const session = await auth()` at the top of the handler.",
        },
        {
            "id": "global_state_tenant",
            "regex": re.compile(r"(globalThis|global)\.(currentTenant|workspace|tenant)\b"),
            "severity": SEVERITY_CRITICAL,
            "message": "Tenant context stored in global state (causes cross-request leaks).",
            "fix": "Pass tenant context through request/session, not global state.",
        },
        {
            "id": "no_rls_hint",
            "regex": re.compile(r"\.execute\(\s*sql`[^`]*`\s*\)(?!.*WHERE)"),
            "severity": SEVERITY_INFO,
            "message": "Raw SQL execution without visible WHERE clause (ensure RLS or manual scoping).",
            "fix": "Verify row-level security is enabled or add explicit tenant filter.",
        },
    ]

    file_count = 0
    for root, _dirs, files in os.walk(src_dir):
        # Skip node_modules and hidden directories
        parts = root.split(os.sep)
        if any(p.startswith(".") or p == "node_modules" for p in parts):
            continue
        for fname in files:
            if not any(fname.endswith(ext) for ext in extensions):
                continue
            fpath = os.path.join(root, fname)
            file_count += 1
            try:
                with open(fpath, "r", errors="replace") as f:
                    lines = f.readlines()
            except OSError:
                continue

            for line_num, line in enumerate(lines, start=1):
                for pat in patterns:
                    if pat["regex"].search(line):
                        rel_path = os.path.relpath(fpath, src_dir)
                        findings.append(finding(
                            pat["id"], pat["severity"],
                            f"{rel_path}:{line_num}: {pat['message']}",
                            fix=pat["fix"],
                            file=rel_path, line=line_num,
                            matched_text=line.strip()[:120],
                        ))

    if file_count == 0:
        findings.append(finding("no_source_files", SEVERITY_INFO,
                                f"No source files found in {src_dir} with extensions {extensions}.",
                                fix="Check the --src path and --extensions flags."))

    return findings


def finding(rule_id, severity, message, fix="", file=None, line=None, matched_text=None):
    """Create a standardized finding dict."""
    f = {"rule_id": rule_id, "severity": severity, "message": message, "fix": fix}
    if file:
        f["file"] = file
    if line is not None:
        f["line"] = line
    if matched_text:
        f["matched_text"] = matched_text
    return f


# ---------------------------------------------------------------------------
# Reporting
# ---------------------------------------------------------------------------

def summarize(findings):
    """Return summary counts by severity."""
    counts = {SEVERITY_CRITICAL: 0, SEVERITY_WARNING: 0, SEVERITY_INFO: 0}
    for f in findings:
        counts[f["severity"]] = counts.get(f["severity"], 0) + 1
    return counts


def print_human_report(findings, title="Tenant Configuration Validation"):
    """Print findings in a human-readable format."""
    print(f"\n{title}")
    print("=" * 70)

    counts = summarize(findings)
    total = len(findings)
    print(f"  Total findings: {total}")
    print(f"  Critical: {counts[SEVERITY_CRITICAL]}  |  Warnings: {counts[SEVERITY_WARNING]}  |  Info: {counts[SEVERITY_INFO]}")
    print("-" * 70)

    if not findings:
        print("  No issues found. Configuration looks good.")
        return

    sorted_findings = sorted(findings, key=lambda f: SEVERITY_ORDER.get(f["severity"], 9))

    for i, f in enumerate(sorted_findings, 1):
        sev = f["severity"].upper()
        print(f"\n  [{sev}] {f['message']}")
        if f.get("matched_text"):
            print(f"          Code: {f['matched_text']}")
        if f.get("fix"):
            print(f"          Fix:  {f['fix']}")

    print("\n" + "-" * 70)
    passed = counts[SEVERITY_CRITICAL] == 0
    verdict = "PASSED (no critical issues)" if passed else "FAILED (critical issues found)"
    print(f"  Verdict: {verdict}")
    print()


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description="Validate multi-tenant configuration for isolation issues and missing tenant scoping.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=textwrap.dedent("""\
            examples:
              %(prog)s --config tenant_config.json
              %(prog)s --src ./app --src ./lib
              %(prog)s --config tenant_config.json --src ./app --json
              %(prog)s --src ./app --extensions .ts .tsx
        """),
    )

    parser.add_argument("--config", default=None,
                        help="Path to tenant configuration JSON file to validate")
    parser.add_argument("--src", action="append", default=None,
                        help="Source directory to scan for isolation issues (repeatable)")
    parser.add_argument("--extensions", nargs="*", default=None,
                        help="File extensions to scan (default: .ts .tsx .js .jsx)")
    parser.add_argument("--severity", choices=["critical", "warning", "info"], default="info",
                        help="Minimum severity to report (default: info)")
    parser.add_argument("--json", action="store_true", dest="json_output",
                        help="Output results as JSON")

    args = parser.parse_args()

    if not args.config and not args.src:
        parser.error("Provide at least one of --config or --src to validate.")

    all_findings = []
    min_sev = SEVERITY_ORDER.get(args.severity, 2)

    # Validate config file
    if args.config:
        config_findings = validate_config_file(args.config)
        all_findings.extend(config_findings)

    # Scan source directories
    if args.src:
        exts = tuple(args.extensions) if args.extensions else None
        for src_dir in args.src:
            src_findings = scan_source_for_issues(src_dir, extensions=exts)
            all_findings.extend(src_findings)

    # Filter by severity
    filtered = [f for f in all_findings if SEVERITY_ORDER.get(f["severity"], 9) <= min_sev]

    counts = summarize(filtered)
    has_critical = counts[SEVERITY_CRITICAL] > 0

    result = {
        "success": not has_critical,
        "total_findings": len(filtered),
        "summary": counts,
        "findings": filtered,
        "validated_at": datetime.now(timezone.utc).isoformat(),
        "config_file": args.config,
        "source_dirs": args.src or [],
    }

    if args.json_output:
        print(json.dumps(result, indent=2))
    else:
        print_human_report(filtered)

    sys.exit(1 if has_critical else 0)


if __name__ == "__main__":
    main()