Build an AI Content Moderation Pipeline

// src/moderation/classifier.ts — Multi-signal content classification
import OpenAI from "openai";
import { Redis } from "ioredis";
import { pool } from "../db";

const openai = new OpenAI();
const redis = new Redis(process.env.REDIS_URL!);

interface ModerationResult {
  contentId: string;
  decision: "approve" | "reject" | "escalate";
  scores: {
    spam: number;
    toxicity: number;
    nsfw: number;
    pii: number;
    violence: number;
  };
  reasons: string[];
  confidence: number;
  processingTimeMs: number;
}

// Thresholds for auto-decisions
const THRESHOLDS = {
  autoReject: 0.9,    // above this = auto-reject
  autoApprove: 0.15,  // below this on ALL categories = auto-approve
  escalate: 0.5,      // between approve and reject = human review
};

export async function moderateContent(
  contentId: string,
  text: string,
  imageUrls: string[] = [],
  userId: string
): Promise<ModerationResult> {
  const startTime = Date.now();
  const reasons: string[] = [];

  // Check user reputation (repeat offenders get stricter thresholds)
  const reputation = await getUserReputation(userId);

  // 1. OpenAI Moderation API (fast, free)
  const moderation = await openai.moderations.create({ input: text });
  const categories = moderation.results[0].category_scores;

  let scores = {
    spam: 0,
    toxicity: Math.max(categories.harassment, categories["harassment/threatening"], categories.hate, categories["hate/threatening"]),
    nsfw: Math.max(categories.sexual, categories["sexual/minors"]),
    pii: 0,
    violence: Math.max(categories.violence, categories["violence/graphic"]),
  };

  // 2. Spam detection (pattern-based + AI)
  scores.spam = await detectSpam(text, userId);

  // 3. PII detection (regex + context)
  scores.pii = detectPII(text);

  // 4. Image moderation (if images present)
  if (imageUrls.length > 0) {
    const imageScores = await moderateImages(imageUrls);
    scores.nsfw = Math.max(scores.nsfw, imageScores.nsfw);
    scores.violence = Math.max(scores.violence, imageScores.violence);
  }

  // Adjust thresholds for low-reputation users
  const rejectThreshold = reputation < 0.3 ? 0.7 : THRESHOLDS.autoReject;
  const approveThreshold = reputation < 0.3 ? 0.1 : THRESHOLDS.autoApprove;

  // Make decision
  let decision: "approve" | "reject" | "escalate" = "approve";
  const maxScore = Math.max(...Object.values(scores));

  if (maxScore > rejectThreshold) {
    decision = "reject";
    reasons.push(...getReasons(scores, rejectThreshold));
  } else if (maxScore > approveThreshold) {
    decision = "escalate";
    reasons.push(...getReasons(scores, approveThreshold));
  }

  const confidence = decision === "approve" ? 1 - maxScore : maxScore;

  const result: ModerationResult = {
    contentId,
    decision,
    scores,
    reasons,
    confidence: Math.round(confidence * 100) / 100,
    processingTimeMs: Date.now() - startTime,
  };

  // Store result
  await pool.query(
    `INSERT INTO moderation_results (content_id, user_id, decision, scores, reasons, confidence, processing_time_ms, created_at)
     VALUES ($1, $2, $3, $4, $5, $6, $7, NOW())`,
    [contentId, userId, decision, JSON.stringify(scores), reasons, confidence, result.processingTimeMs]
  );

  // Apply decision
  if (decision === "reject") {
    await pool.query("UPDATE posts SET status = 'rejected', moderation_reason = $2 WHERE id = $1", [contentId, reasons.join(", ")]);
    await updateUserReputation(userId, -0.1);
  } else if (decision === "escalate") {
    await redis.rpush("moderation:queue", JSON.stringify({ contentId, userId, scores, reasons }));
  } else {
    await pool.query("UPDATE posts SET status = 'approved' WHERE id = $1", [contentId]);
  }

  return result;
}

async function detectSpam(text: string, userId: string): Promise<number> {
  let score = 0;

  // Pattern checks
  const urlCount = (text.match(/https?:\/\//g) || []).length;
  if (urlCount > 3) score += 0.3;

  // Excessive caps
  const capsRatio = (text.match(/[A-Z]/g) || []).length / Math.max(text.length, 1);
  if (capsRatio > 0.7 && text.length > 20) score += 0.2;

  // Repeated characters
  if (/(.)\1{5,}/.test(text)) score += 0.2;

  // Posting frequency (rate-based spam detection)
  const recentPosts = await redis.incr(`spam:rate:${userId}`);
  await redis.expire(`spam:rate:${userId}`, 300);
  if (recentPosts > 10) score += 0.4; // 10+ posts in 5 min

  return Math.min(1, score);
}

function detectPII(text: string): number {
  let score = 0;

  // Email addresses
  if (/[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}/.test(text)) score += 0.5;

  // Phone numbers
  if (/\b\d{3}[-.]?\d{3}[-.]?\d{4}\b/.test(text)) score += 0.5;

  // SSN patterns
  if (/\b\d{3}-\d{2}-\d{4}\b/.test(text)) score += 0.9;

  // Credit card patterns
  if (/\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b/.test(text)) score += 0.9;

  return Math.min(1, score);
}

async function moderateImages(urls: string[]): Promise<{ nsfw: number; violence: number }> {
  let maxNsfw = 0, maxViolence = 0;

  for (const url of urls.slice(0, 5)) {
    const response = await openai.chat.completions.create({
      model: "gpt-4o-mini",
      messages: [{
        role: "user",
        content: [
          { type: "text", text: "Rate this image 0-1 for NSFW and violence content. Reply JSON: {nsfw: 0.0, violence: 0.0}" },
          { type: "image_url", image_url: { url } },
        ],
      }],
      max_tokens: 50,
    });

    try {
      const parsed = JSON.parse(response.choices[0].message.content || "{}");
      maxNsfw = Math.max(maxNsfw, parsed.nsfw || 0);
      maxViolence = Math.max(maxViolence, parsed.violence || 0);
    } catch { /* unparseable response */ }
  }

  return { nsfw: maxNsfw, violence: maxViolence };
}

function getReasons(scores: Record<string, number>, threshold: number): string[] {
  return Object.entries(scores)
    .filter(([_, score]) => score > threshold)
    .map(([category, score]) => `${category}: ${(score * 100).toFixed(0)}%`);
}

async function getUserReputation(userId: string): Promise<number> {
  const cached = await redis.get(`reputation:${userId}`);
  if (cached) return parseFloat(cached);

  const { rows } = await pool.query(
    "SELECT reputation_score FROM user_profiles WHERE user_id = $1",
    [userId]
  );
  const score = rows[0]?.reputation_score ?? 1.0;
  await redis.setex(`reputation:${userId}`, 3600, String(score));
  return score;
}

async function updateUserReputation(userId: string, delta: number): Promise<void> {
  await pool.query(
    "UPDATE user_profiles SET reputation_score = GREATEST(0, LEAST(1, reputation_score + $2)) WHERE user_id = $1",
    [userId, delta]
  );
  await redis.del(`reputation:${userId}`);
}