Anatomia do Vivo de Codigo: Como Construi um Blog com Stack de Startup

graph TB
    subgraph "Frontend (Vercel)"
        A[Next.js 16 + React 19]
        A --> B[Tailwind CSS 4]
        A --> C[Server Components]
        A --> D[6 Cron Jobs Proxy]
    end
    
    subgraph "Database (Supabase)"
        E[PostgreSQL + pgvector]
        E --> F[Embeddings 384-dim]
        E --> G[Knowledge Graph]
        E --> H[Analytics Data]
    end
    
    subgraph "VPS Oracle Cloud (137.131.xxx.xxx)"
        I[10 Docker Containers]
        I --> J[ACI - FastAPI]
        I --> K[MinIO Storage]
        I --> L[12 Python Crons]
        I --> M[Evolution API WhatsApp]
    end
    
    subgraph "External Services"
        N[Cloudflare CDN]
        O[Algolia Search]
        P[Resend Email]
        Q[Google AdSense]
        R[Umami Analytics]
    end
    
    A --> E
    A --> I
    A --> N
    A --> O
    J --> E
    K --> N
    D --> I
    L --> E

-- Busca semântica real
SELECT title, content, 
       1 - (embedding <=> query_embedding) as similarity
FROM posts 
ORDER BY embedding <=> query_embedding 
LIMIT 10;

// Antes: Supabase Storage
const { data } = await supabase.storage
  .from('images')
  .upload(fileName, file);
 
// Depois: MinIO + CDN
const imageUrl = await uploadToMinio(file);
const cdnUrl = `https://cdn.vivodecodigo.com.br/${imageUrl}`;

CREATE TABLE entities (
  id UUID PRIMARY KEY,
  name TEXT NOT NULL,
  type entity_type,
  embedding vector(384)
);
 
CREATE TABLE relationships (
  source_id UUID REFERENCES entities(id),
  target_id UUID REFERENCES entities(id),
  relation_type TEXT,
  confidence FLOAT
);

sequenceDiagram
    participant Cron as Cron Job
    participant ACI as ACI System
    participant Claude as Claude Opus 4.6
    participant Reviewer as Auto Reviewer
    participant DB as Database
    participant Blog as Blog Frontend
    
    Cron->>ACI: selectTopic()
    ACI->>ACI: Analyze trends, gaps, competition
    ACI->>Claude: generateDraft(topic, context)
    Claude->>Reviewer: Draft content
    Reviewer->>Reviewer: Flesch score, NLP, Burstiness
    alt Score > 85
        Reviewer->>DB: Save as published
        DB->>Blog: Trigger revalidation
    else Score < 85
        Reviewer->>Claude: Regenerate with feedback
    end

function kmeans(data, k = 3) {
  let centroids = initializeCentroids(data, k);
  let clusters = [];
  
  for (let i = 0; i < 100; i++) {
    clusters = assignToClusters(data, centroids);
    const newCentroids = updateCentroids(clusters);
    
    if (convergence(centroids, newCentroids)) break;
    centroids = newCentroids;
  }
  
  return { clusters, centroids };
}

function predictCTR(features) {
  // features: [title_length, readability_score, hour_published, day_of_week]
  const weights = [0.023, 0.156, -0.089, 0.034]; // Treinados com 27 posts
  const bias = 0.067;
  
  return features.reduce((sum, feature, i) => 
    sum + feature * weights[i], bias
  );
}

-- Tipos de entidades
CREATE TYPE entity_type AS ENUM (
  'technology', 'concept', 'tool', 'framework', 
  'language', 'pattern', 'company', 'person'
);
 
-- Busca por relacionamentos
WITH RECURSIVE related_entities AS (
  -- Anchor: entidade inicial
  SELECT e.id, e.name, e.type, 0 as depth
  FROM entities e 
  WHERE e.name ILIKE '%react%'
  
  UNION ALL
  
  -- Recursão: entidades relacionadas
  SELECT e.id, e.name, e.type, re.depth + 1
  FROM entities e
  JOIN relationships r ON (e.id = r.target_id OR e.id = r.source_id)
  JOIN related_entities re ON (
    (r.source_id = re.id AND e.id = r.target_id) OR
    (r.target_id = re.id AND e.id = r.source_id)
  )
  WHERE re.depth < 3
)
SELECT * FROM related_entities;

# docker-compose.yml simplificado
services:
  aci-api:
    image: fastapi:latest
    ports: ["8000:8000"]
    
  minio:
    image: minio/minio:latest
    ports: ["9000:9000", "9001:9001"]
    
  redis:
    image: redis:alpine
    
  evolution-api:
    image: atendai/evolution-api:latest
    
  nginx:
    image: nginx:alpine
    ports: ["80:80", "443:443"]
    
  # + 5 outros serviços

docker stats --no-stream --format '{{json .}}' vdc-aci vdc-minio vdc-redis vdc-caddy

class CircuitBreaker {
  constructor(threshold = 3, timeout = 30000) {
    this.failureCount = 0;
    this.threshold = threshold;
    this.timeout = timeout;
    this.state = 'CLOSED'; // CLOSED, OPEN, HALF_OPEN
    this.nextAttempt = Date.now();
  }
  
  async execute(operation) {
    if (this.state === 'OPEN') {
      if (Date.now() < this.nextAttempt) {
        throw new Error('Circuit breaker is OPEN');
      }
      this.state = 'HALF_OPEN';
    }
    
    try {
      const result = await operation();
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure();
      throw error;
    }
  }
  
  onSuccess() {
    this.failureCount = 0;
    this.state = 'CLOSED';
  }
  
  onFailure() {
    this.failureCount++;
    if (this.failureCount >= this.threshold) {
      this.state = 'OPEN';
      this.nextAttempt = Date.now() + this.timeout;
      this.sendWhatsAppAlert();
    }
  }
  
  sendWhatsAppAlert() {
    // Evolution API para alertas
    fetch('http://137.131.xxx.xxx:8080/message/sendText', {
      method: 'POST',
      body: JSON.stringify({
        number: '5511999999999',
        text: `🚨 Circuit breaker OPEN: ${this.service}`
      })
    });
  }
}

import { Plate, createPlugins } from '@udecode/plate';
 
const plugins = createPlugins([
  // Rich text
  createBoldPlugin(),
  createItalicPlugin(),
  createCodePlugin(),
  
  // Links
  createLinkPlugin(),
  
  // Lists
  createListPlugin(),
  
  // Code blocks
  createCodeBlockPlugin({
    options: {
      prism: {
        languages: ['javascript', 'typescript', 'python', 'sql']
      }
    }
  })
]);
 
export function CommentEditor({ onSubmit }) {
  return (
    <Plate plugins={plugins}>
      <PlateEditor placeholder="Escreva seu comentário..." />
      <Toolbar>
        <BoldToolbarButton />
        <ItalicToolbarButton />
        <CodeToolbarButton />
        <LinkToolbarButton />
      </Toolbar>
    </Plate>
  );
}

import { Turnstile } from '@marsidev/react-turnstile';
 
export function CommentForm() {
  const [turnstileToken, setTurnstileToken] = useState('');
  
  const handleSubmit = async (data) => {
    if (!turnstileToken) {
      toast.error('Complete a verificação de segurança');
      return;
    }
    
    await submitComment({ ...data, turnstileToken });
  };
  
  return (
    <form onSubmit={handleSubmit}>
      <CommentEditor />
      <Turnstile
        siteKey="0x4AAAAAAA..."
        onSuccess={setTurnstileToken}
      />
      <button type="submit">Publicar Comentário</button>
    </form>
  );
}

// Server Component
export async function RevenueRouter({ userId, postId }) {
  const userSegment = await getUserSegment(userId);
  const postCategory = await getPostCategory(postId);
  
  const adConfig = {
    'backend-expert': {
      slots: ['responsive-top', 'in-article', 'sidebar'],
      keywords: ['database', 'api', 'performance']
    },
    'frontend-enthusiast': {
      slots: ['responsive-top', 'in-article'],
      keywords: ['react', 'nextjs', 'css']
    },
    'beginner': {
      slots: ['responsive-top'],
      keywords: ['tutorial', 'guide', 'basics']
    }
  };
  
  const config = adConfig[userSegment] || adConfig['beginner'];
  
  return (
    <div className="ad-container">
      {config.slots.map(slot => (
        <GoogleAd
          key={slot}
          slot={slot}
          keywords={config.keywords}
          format="auto"
        />
      ))}
    </div>
  );
}

from prophet import Prophet
from statsmodels.tsa.arima.model import ARIMA
import pandas as pd
 
class TrendForecaster:
    def __init__(self):
        self.signals = [
            'google_trends', 'github_stars', 'npm_downloads',
            'stack_overflow_questions', 'job_postings',
            'twitter_mentions', 'reddit_discussions'
        ]
    
    def forecast_technology_trend(self, tech_name, days=30):
        # Coleta dados históricos
        data = self.collect_signals(tech_name)
        
        # Prophet para tendências com sazonalidade
        prophet_forecast = self.prophet_forecast(data, days)
        
        # ARIMA para padrões lineares
        arima_forecast = self.arima_forecast(data, days)
        
        # Ensemble dos dois modelos
        final_forecast = self.ensemble_forecast(
            prophet_forecast, 
            arima_forecast
        )
        
        return {
            'trend_direction': final_forecast['direction'],
            'confidence': final_forecast['confidence'],
            'peak_date': final_forecast['peak'],
            'signals_strength': final_forecast['signals']
        }
    
    def prophet_forecast(self, data, days):
        model = Prophet(
            changepoint_prior_scale=0.05,
            seasonality_prior_scale=10,
            holidays_prior_scale=10
        )
        
        df = pd.DataFrame({
            'ds': data['dates'],
            'y': data['values']
        })
        
        model.fit(df)
        
        future = model.make_future_dataframe(periods=days)
        forecast = model.predict(future)
        
        return forecast

import feedparser
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
 
# Lista curta para demonstrar a API; revisar para o corpus utilizado.
STOP_WORDS_PT = [
    "de", "do", "da", "dos", "das", "em",
    "para", "por", "com", "um", "uma",
]
 
class CompetitiveIntelligence:
    def __init__(self):
        self.competitors = [
            'https://blog.rocketseat.com.br/feed/',
            'https://willianjusten.com.br/feed.xml',
            'https://tableless.com.br/feed/',
            'https://imasters.com.br/feed/',
            'https://medium.com/feed/@programador-real'
        ]
        
    def analyze_content_gaps(self):
        # Coleta posts dos concorrentes
        competitor_posts = []
        for feed_url in self.competitors:
            posts = self.fetch_feed(feed_url)
            competitor_posts.extend(posts)
        
        # Meus posts existentes
        my_posts = self.fetch_my_posts()
        
        # Análise TF-IDF para encontrar gaps
        all_content = [p['content'] for p in competitor_posts + my_posts]
        
        vectorizer = TfidfVectorizer(
            max_features=1000,
            stop_words=STOP_WORDS_PT,
            ngram_range=(1, 2),
        )
        
        tfidf_matrix = vectorizer.fit_transform(all_content)
        
        # Identifica tópicos que concorrentes cobrem mas eu não
        competitor_matrix = tfidf_matrix[:len(competitor_posts)]
        my_matrix = tfidf_matrix[len(competitor_posts):]
        
        gaps = self.find_content_gaps(
            competitor_matrix, 
            my_matrix, 
            vectorizer.get_feature_names_out()
        )
        
        return gaps
    
    def find_content_gaps(self, competitor_matrix, my_matrix, features):
        # Média TF-IDF dos concorrentes por feature
        competitor_means = competitor_matrix.mean(axis=0).A1
        
        # Média TF-IDF dos meus posts por feature
        my_means = my_matrix.mean(axis=0).A1 if my_matrix.shape[0] > 0 else np.zeros(len(features))
        
        # Gap = alta relevância nos concorrentes, baixa nos meus
        gaps = []
        for i, feature in enumerate(features):
            if competitor_means[i] > 0.05 and my_means[i] < 0.02:
                gaps.append({
                    'topic': feature,
                    'competitor_relevance': competitor_means[i],
                    'my_relevance': my_means[i],
                    'gap_score': competitor_means[i] - my_means[i]
                })
        
        return sorted(gaps, key=lambda x: x['gap_score'], reverse=True)

// /api/crons/content-pipeline.js
export default async function handler(req, res) {
  if (req.method !== 'POST') {
    return res.status(405).json({ error: 'Method not allowed' });
  }
  
  const authHeader = req.headers.authorization;
  if (authHeader !== `Bearer ${process.env.CRON_SECRET}`) {
    return res.status(401).json({ error: 'Unauthorized' });
  }
  
  try {
    // Proxy para VPS
    const response = await fetch('http://137.131.xxx.xxx:8000/pipeline/run', {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${process.env.ACI_TOKEN}`,
        'Content-Type': 'application/json'
      }
    });
    
    const result = await response.json();
    
    res.status(200).json({
      success: true,
      pipeline_result: result
    });
  } catch (error) {
    res.status(500).json({
      success: false,
      error: error.message
    });
  }
}

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
import asyncio
 
class CronManager:
    def __init__(self):
        self.scheduler = AsyncIOScheduler()
        self.setup_jobs()
    
    def setup_jobs(self):
        # Content pipeline - a cada 6 horas
        self.scheduler.add_job(
            self.run_content_pipeline,
            CronTrigger(hour='*/6'),
            id='content_pipeline'
        )
        
        # Trend analysis - diário às 2h
        self.scheduler.add_job(
            self.analyze_trends,
            CronTrigger(hour=2),
            id='trend_analysis'
        )
        
        # Competitive intelligence - diário às 4h
        self.scheduler.add_job(
            self.update_competitive_data,
            CronTrigger(hour=4),
            id='competitive_intel'
        )
        
        # Knowledge graph update - a cada 2 horas
        self.scheduler.add_job(
            self.update_knowledge_graph,
            CronTrigger(hour='*/2'),
            id='kg_update'
        )
        
        # Analytics processing - a cada hora
        self.scheduler.add_job(
            self.process_analytics,
            CronTrigger(minute=0),
            id='analytics'
        )
    
    def start(self):
        self.scheduler.start()
        print("✅ Cron jobs iniciados")

import logging
from datetime import datetime
 
class CronMonitor:
    def __init__(self):
        self.execution_log = []
    
    def log_execution(self, job_id, status, duration, error=None):
        log_entry = {
            'job_id': job_id,
            'timestamp': datetime.now(),
            'status': status,
            'duration': duration,
            'error': error
        }
        
        self.execution_log.append(log_entry)
        
        # Alerta se job falhar 3 vezes seguidas
        recent_failures = [
            log for log in self.execution_log[-10:]
            if log['job_id'] == job_id and log['status'] == 'failed'
        ]
        
        if len(recent_failures) >= 3:
            self.send_whatsapp_alert(f"🚨 Job {job_id} falhando consistentemente")

Oracle Cloud VPS: $0.00 (Always Free)
Supabase: $0.00 (Free tier: 500MB DB, 1GB transfer)
Vercel: $0.00 (Hobby plan: 100GB bandwidth)
Cloudflare: $0.00 (Free tier: CDN unlimited)
Algolia: $0.00 (10K searches/mês)
Resend: $0.00 (3K emails/mês)
Google AdSense: $0.00 (revenue share)
Umami: $0.00 (self-hosted)
 
Total mensal: $0.00

export async function hybridSearch(query, limit = 10) {
  // Busca tradicional no Algolia
  const algoliaResults = await algoliaClient.search(query, {
    hitsPerPage: limit * 2,
    attributesToRetrieve: ['title', 'excerpt', 'slug', 'objectID']
  });
  
  // Busca semântica no pgvector
  const queryEmbedding = await generateEmbedding(query);
  
  const { data: semanticResults } = await supabase.rpc('semantic_search', {
    query_embedding: queryEmbedding,
    match_threshold: 0.7,
    match_count: limit * 2
  });
  
  // Combina e ranqueia resultados
  const combinedResults = combineSearchResults(
    algoliaResults.hits,
    semanticResults,
    query
  );
  
  return combinedResults.slice(0, limit);
}
 
function combineSearchResults(algoliaHits, semanticHits, query) {
  const results = new Map();
  
  // Adiciona resultados do Algolia com peso
  algoliaHits.forEach((hit, index) => {
    const score = (algoliaHits.length - index) / algoliaHits.length;
    results.set(hit.objectID, {
      ...hit,
      algolia_score: score,
      semantic_score: 0,
      combined_score: score * 0.6 // Peso 60% para keyword search
    });
  });
  
  // Adiciona/atualiza com resultados semânticos
  semanticHits.forEach((hit, index) => {
    const score = hit.similarity;
    const existing = results.get(hit.id);
    
    if (existing) {
      // Combina scores se resultado existe em ambos
      existing.semantic_score = score;
      existing.combined_score = existing.algolia_score * 0.6 + score * 0.4;
    } else {
      // Adiciona resultado apenas semântico
      results.set(hit.id, {
        ...hit,
        algolia_score: 0,
        semantic_score: score,
        combined_score: score * 0.4 // Peso 40% para semantic search
      });
    }
  });
  
  // Ordena por score combinado
  return Array.from(results.values())
    .sort((a, b) => b.combined_score - a.combined_score);
}

CREATE OR REPLACE FUNCTION semantic_search(
  query_embedding vector(384),
  match_threshold float,
  match_count int
)
RETURNS TABLE (
  id uuid,
  title text,
  excerpt text,
  slug text,
  similarity float
)
LANGUAGE plpgsql
AS $$
BEGIN
  RETURN QUERY
  SELECT
    posts.id,
    posts.title,
    posts.excerpt,
    posts.slug,
    1 - (posts.embedding <=> query_embedding) as similarity
  FROM posts
  WHERE 1 - (posts.embedding <=> query_embedding) > match_threshold
  ORDER BY posts.embedding <=> query_embedding
  LIMIT match_count;
END;
$$;

class ContentRefreshAnalyzer:
    def __init__(self):
        self.signals = [
            'technology_evolution',
            'external_link_decay', 
            'performance_decline',
            'competitive_updates',
            'user_feedback'
        ]
    
    def analyze_post_freshness(self, post_id):
        scores = {}
        
        # 1. Technology Evolution Score
        tech_mentions = self.extract_technologies(post_id)
        evolution_score = 0
        for tech in tech_mentions:
            days_since_update = self.days_since_tech_update(tech)
            if days_since_update > 90:  # 3 meses
                evolution_score += 1
        scores['tech_evolution'] = evolution_score / len(tech_mentions) if tech_mentions else 0
        
        # 2. External Link Decay
        external_links = self.get_external_links(post_id)
        broken_links = 0
        for link in external_links:
            if not self.is_link_valid(link):
                broken_links += 1
        scores['link_decay'] = broken_links / len(external_links) if external_links else 0
        
        # 3. Performance Decline
        current_metrics = self.get_post_metrics(post_id, days=30)
        historical_metrics = self.get_post_metrics(post_id, days=90, offset=90)
        
        performance_decline = 0
        if historical_metrics['avg_time_on_page'] > 0:
            performance_decline = 1 - (current_metrics['avg_time_on_page'] / historical_metrics['avg_time_on_page'])
        scores['performance_decline'] = max(0, performance_decline)
        
        # 4. Competitive Updates
        similar_posts = self.find_competitor_posts(post_id)
        recent_competitor_updates = len([
            p for p in similar_posts 
            if p['updated_at'] > (datetime.now() - timedelta(days=60))
        ])
        scores['competitive_updates'] = min(1.0, recent_competitor_updates / 5)
        
        # 5. User Feedback
        comments = self.get_post_comments(post_id)
        outdated_mentions = len([
            c for c in comments 
            if any(word in c['content'].lower() for word in ['desatualizado', 'antigo', 'mudou', 'não funciona'])
        ])
        scores['user_feedback'] = min(1.0, outdated_mentions / 3)
        
        # Score final ponderado
        weights = {
            'tech_evolution': 0.3,
            'link_decay': 0.2,
            'performance_decline': 0.2,
            'competitive_updates': 0.15,
            'user_feedback': 0.15
        }
        
        final_score = sum(scores[signal] * weights[signal] for signal in scores)
        
        return {
            'refresh_score': final_score,
            'signals': scores,
            'recommendation': self.get_refresh_recommendation(final_score),
            'priority': self.get_refresh_priority(final_score)
        }
    
    def get_refresh_recommendation(self, score):
        if score > 0.7:
            return "URGENT_REFRESH"
        elif score > 0.5:
            return "SCHEDULED_REFRESH"
        elif score > 0.3:
            return "MONITOR"
        else:
            return "FRESH"

from fastapi import FastAPI, BackgroundTasks
from typing import List, Dict
import asyncio
 
app = FastAPI(title="ACI - Autonomous Content Intelligence")
 
class ACISystem:
    def __init__(self):
        self.modules = {
            'intent': IntentAnalyzer(),
            'consent': ConsentManager(), 
            'trends': TrendAnalyzer(),
            'kg': KnowledgeGraphManager(),
            'competitive': CompetitiveIntelligence(),
            'sentiment': SentimentAnalyzer(),
            'performance': PerformancePredictor(),
            'content': ContentGenerator(),
            'seo': SEOOptimizer(),
            'social': SocialMediaManager(),
            'email': EmailCampaignManager(),
            'analytics': AnalyticsProcessor()
        }
    
    async def orchestrate_content_pipeline(self):
        # 1. Analyze current trends and gaps
        trends = await self.modules['trends'].get_trending_topics()
        gaps = await self.modules['competitive'].analyze_content_gaps()
        
        # 2. Select best topic using multiple signals
        selected_topic = await self.select_optimal_topic(trends, gaps)
        
        # 3. Generate content with context
        content = await self.modules['content'].generate_post(
            topic=selected_topic,
            context=await self.build_context(selected_topic)
        )
        
        # 4. Optimize for SEO
        optimized_content = await self.modules['seo'].optimize(content)
        
        # 5. Predict performance
        performance_prediction = await self.modules['performance'].predict(
            optimized_content
        )
        
        # 6. Auto-publish if quality threshold met
        if performance_prediction['quality_score'] > 0.85:
            await self.publish_content(optimized_content)
            await self.modules['social'].schedule_promotion(optimized_content)
            
        return {
            'status': 'completed',
            'topic': selected_topic,
            'quality_score': performance_prediction['quality_score'],
            'published': performance_prediction['quality_score'] > 0.85
        }
 
@app.post("/pipeline/run")
async def run_content_pipeline(background_tasks: BackgroundTasks):
    aci = ACISystem()
    background_tasks.add_task(aci.orchestrate_content_pipeline)
    
    return {"status": "Pipeline initiated", "timestamp": datetime.now()}
 
@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "modules": len(ACISystem().modules),
        "uptime": get_uptime()
    }