Claude's Corner: Grade, The Payroll API Built for a World That Pays on Results

# Build Guide: Clone Grade, Performance-Based Payroll API

**Stack**: Next.js 14, FastAPI (Python), PostgreSQL, Stripe, Redis, AWS

---

## Step 1: Database Schema

```sql
-- Workers / payees
CREATE TABLE workers (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  company_id UUID NOT NULL REFERENCES companies(id),
  external_id TEXT NOT NULL,              -- your system's contractor ID
  name TEXT NOT NULL,
  email TEXT NOT NULL,
  country_code CHAR(2) NOT NULL,
  tax_id TEXT,                            -- encrypted at rest via pgcrypto
  tax_form_type TEXT,                     -- '1099-NEC', 'W-8BEN', etc.
  stripe_account_id TEXT,                 -- for payouts via Stripe Connect
  status TEXT NOT NULL DEFAULT 'active',
  created_at TIMESTAMPTZ DEFAULT now(),
  UNIQUE (company_id, external_id)
);

-- Payout rules (the configurable rule engine)
CREATE TABLE payout_rules (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  company_id UUID NOT NULL REFERENCES companies(id),
  name TEXT NOT NULL,
  event_type TEXT NOT NULL,               -- e.g. 'video_published', 'invoice_processed'
  conditions JSONB,                       -- e.g. {"views": {">": 10000}}
  payout_type TEXT NOT NULL,             -- 'flat', 'per_unit', 'tiered'
  payout_amount_cents BIGINT,
  payout_tiers JSONB,                    -- [{min: 0, max: 1000, rate_cents: 10}, ...]
  currency TEXT NOT NULL DEFAULT 'usd',
  active BOOLEAN DEFAULT true,
  created_at TIMESTAMPTZ DEFAULT now()
);

-- Performance events ingested via API/webhook
CREATE TABLE performance_events (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  company_id UUID NOT NULL,
  worker_id UUID NOT NULL REFERENCES workers(id),
  idempotency_key TEXT NOT NULL,          -- client-supplied, deduplicated
  event_type TEXT NOT NULL,
  quantity NUMERIC NOT NULL DEFAULT 1,
  metadata JSONB,
  processed BOOLEAN DEFAULT false,
  payout_id UUID,                         -- set once matched to a payout
  created_at TIMESTAMPTZ DEFAULT now(),
  UNIQUE (company_id, idempotency_key)
);

-- Calculated payouts (pending approval)
CREATE TABLE payouts (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  company_id UUID NOT NULL,
  worker_id UUID NOT NULL REFERENCES workers(id),
  payout_rule_id UUID REFERENCES payout_rules(id),
  amount_cents BIGINT NOT NULL,
  currency TEXT NOT NULL DEFAULT 'usd',
  status TEXT NOT NULL DEFAULT 'pending', -- pending, approved, processing, paid, failed
  period_start DATE,
  period_end DATE,
  stripe_transfer_id TEXT,
  error_message TEXT,
  approved_by UUID,
  approved_at TIMESTAMPTZ,
  paid_at TIMESTAMPTZ,
  created_at TIMESTAMPTZ DEFAULT now(),
  updated_at TIMESTAMPTZ DEFAULT now()
);

-- Annual tax reporting tracker
CREATE TABLE tax_filings (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  company_id UUID NOT NULL,
  worker_id UUID NOT NULL,
  tax_year INT NOT NULL,
  form_type TEXT NOT NULL,
  total_paid_cents BIGINT NOT NULL,
  filing_status TEXT DEFAULT 'pending',  -- pending, filed, corrected
  filed_at TIMESTAMPTZ,
  UNIQUE (company_id, worker_id, tax_year, form_type)
);
```

---

## Step 2: API Design

**Event Ingestion** (idempotent):
```
POST /v1/events
Authorization: Bearer {api_key}
{
  "idempotency_key": "vid_123456_published",
  "worker_id": "wrk_abc",
  "event_type": "video_published",
  "quantity": 1,
  "metadata": {"views": 45000, "platform": "youtube"}
}
```

**Rule Engine Evaluation** (background job):
- On new event insert, push to Redis queue
- Worker picks up, queries matching payout rules for company
- Evaluates conditions (jsonpath or custom evaluator)
- Creates `payouts` row if rule matches, links event → payout

**Payout Approval** (human-in-the-loop):
```
GET  /v1/payouts?status=pending&period=2026-06
POST /v1/payouts/{id}/approve
POST /v1/payouts/batch-approve  { "payout_ids": [...] }
```

**Disbursement** (triggered on approval):
```
POST /v1/payouts/{id}/disburse  → triggers Stripe Transfer
```

**Worker Onboarding**:
```
POST /v1/workers          → create payee
POST /v1/workers/{id}/kyc → trigger Stripe Connect onboarding link
GET  /v1/workers/{id}/tax-forms?year=2025
```

---

## Step 3: Rule Engine (Key Algorithm)

```python
import jsonschema
from decimal import Decimal

def evaluate_rule(rule: dict, event: dict) -> int | None:
    """Returns payout amount in cents, or None if rule doesn't match."""
    conditions = rule.get("conditions", {})
    metadata = event.get("metadata", {})

    for field, constraint in conditions.items():
        value = metadata.get(field)
        if value is None:
            return None
        for op, threshold in constraint.items():
            if op == ">" and not (value > threshold): return None
            if op == ">=" and not (value >= threshold): return None
            if op == "<" and not (value < threshold): return None
            if op == "==" and not (value == threshold): return None

    qty = Decimal(event.get("quantity", 1))

    if rule["payout_type"] == "flat":
        return rule["payout_amount_cents"]
    elif rule["payout_type"] == "per_unit":
        return int(qty * rule["payout_amount_cents"])
    elif rule["payout_type"] == "tiered":
        for tier in sorted(rule["payout_tiers"], key=lambda t: t["min"]):
            if tier["min"] <= qty <= tier.get("max", float("inf")):
                return int(qty * tier["rate_cents"])
    return None
```

---

## Step 4: Idempotency & Deduplication

```python
# FastAPI endpoint with idempotency key check
@router.post("/v1/events")
async def ingest_event(event: EventSchema, db: AsyncSession = Depends(get_db)):
    # Idempotency: return existing if key already processed
    existing = await db.execute(
        select(PerformanceEvent).where(
            PerformanceEvent.idempotency_key == event.idempotency_key,
            PerformanceEvent.company_id == event.company_id
        )
    )
    if existing.scalar_one_or_none():
        return {"status": "duplicate", "message": "Event already received"}

    new_event = PerformanceEvent(**event.dict())
    db.add(new_event)
    await db.commit()

    # Enqueue for rule evaluation
    await redis.rpush("events:pending", str(new_event.id))
    return {"status": "accepted", "event_id": str(new_event.id)}
```

---

## Step 5: Stripe Connect for Global Payouts

```python
import stripe

def onboard_worker(worker_id: str, email: str, country: str) -> str:
    """Create Stripe Connect Express account and return onboarding URL."""
    account = stripe.Account.create(
        type="express",
        country=country,
        email=email,
        capabilities={"transfers": {"requested": True}},
    )
    # Store account.id on workers row
    link = stripe.AccountLink.create(
        account=account.id,
        refresh_url=f"https://app.usegrade.com/workers/{worker_id}/retry",
        return_url=f"https://app.usegrade.com/workers/{worker_id}/onboarded",
        type="account_onboarding",
    )
    return link.url

def disburse_payout(payout_id: str, amount_cents: int,
                    stripe_account_id: str, currency: str = "usd") -> str:
    transfer = stripe.Transfer.create(
        amount=amount_cents,
        currency=currency,
        destination=stripe_account_id,
        metadata={"payout_id": payout_id},
    )
    return transfer.id
```

---

## Step 6: Tax Compliance Automation

```python
# Year-end 1099 aggregation job (run January 1)
async def generate_1099s(company_id: str, tax_year: int):
    result = await db.execute("""
        SELECT worker_id, SUM(amount_cents) as total_cents
        FROM payouts
        WHERE company_id = :company_id
          AND status = 'paid'
          AND EXTRACT(YEAR FROM paid_at) = :year
        GROUP BY worker_id
        HAVING SUM(amount_cents) >= 60000  -- $600 threshold
    """, {"company_id": company_id, "year": tax_year})

    for row in result:
        worker = await get_worker(row.worker_id)
        if worker.country_code == "US":
            form_type = "1099-NEC"
        else:
            form_type = "W-8BEN"
        # Use Stripe Tax or Taxjar API to file, or generate PDF
        await create_tax_filing(company_id, worker.id, tax_year, form_type,
                                row.total_cents)
```

---

## Step 7: Deployment & SOC 2 Readiness

```yaml
# docker-compose.yml (dev)
services:
  api:
    build: ./api
    environment:
      DATABASE_URL: postgresql+asyncpg://postgres:pass@db/grade
      REDIS_URL: redis://redis:6379
      STRIPE_SECRET_KEY: ${STRIPE_SECRET_KEY}
    ports: ["8000:8000"]
  worker:
    build: ./api
    command: python -m workers.rule_evaluator
  db:
    image: postgres:16
    volumes: ["pgdata:/var/lib/postgresql/data"]
  redis:
    image: redis:7-alpine
```

**SOC 2 checklist (start early):**
- Encrypt PII at rest (pgcrypto for tax IDs, SSNs)
- TLS 1.2+ everywhere, HSTS headers
- Immutable audit log table (no UPDATE/DELETE, append-only via trigger)
- MFA enforced on all internal accounts
- Automated access reviews quarterly
- Incident response runbook in place
- Use Vanta or Drata to automate evidence collection, saves ~200 hours of manual prep

**Estimated build time to MVP:** 6, 8 weeks for a senior engineer
**Time to enterprise-ready (SOC 2 + banking approvals):** 9, 14 months

**Key dependencies:**
- Stripe Connect (payouts), apply for platform approval early, takes 2, 4 weeks
- Column.com or Mercury for ACH origination
- Taxjar or Stripe Tax for US state tax calculations
- Vanta for SOC 2 automation