CampaignMetricsService — gRPC

// DataFlow AI — CampaignMetricsService
// Contrato gRPC para el pipeline de métricas de campañas
syntax = "proto3";
package dataflowai.metrics.v1;

import "google/protobuf/timestamp.proto";
import "google/protobuf/empty.proto";

// ─── Servicio principal ───────────────────────────────────────────────────
service CampaignMetricsService {
  // Unary: registra una métrica puntual
  rpc RecordMetric(CampaignMetric) returns (RecordResponse);

  // Unary: devuelve el resumen consolidado de una campaña
  rpc GetCampaignSummary(SummaryRequest) returns (CampaignSummary);

  // Client streaming: ingesta masiva desde ad networks
  rpc BatchIngestMetrics(stream CampaignMetric) returns (IngestSummary);

  // Server streaming: push de alertas de anomalías al dashboard
  rpc StreamAlerts(AlertSubscription) returns (stream Alert);

  // Bidireccional: drill-down interactivo en vivo
  rpc LiveMetricsChat(stream MetricQuery) returns (stream MetricResult);
}

// ─── Mensajes principales ─────────────────────────────────────────────────
message CampaignMetric {
  string   campaign_id  = 1;
  Channel  channel      = 2;  // META | GOOGLE | TIKTOK
  int64    impressions  = 3;
  int64    clicks       = 4;
  double   spend_eur   = 5;
  int32    conversions = 6;
  google.protobuf.Timestamp timestamp = 7;
}

message CampaignSummary {
  string         campaign_id   = 1;
  string         name          = 2;
  double         total_spend_eur = 3;
  double         roas          = 4;  // return on ad spend
  double         ctr_pct      = 5;
  int32          conversions  = 6;
  CampaignStatus status       = 7;
}

message Alert {
  string        campaign_id  = 1;
  AlertSeverity severity    = 2;
  string        message     = 3;
  string        metric_name = 4;
  double        value       = 5;
  double        threshold   = 6;
  google.protobuf.Timestamp fired_at = 7;
}

message MetricQuery  { string campaign_id = 1; Channel channel = 2; string metric = 3; }
message MetricResult { string campaign_id = 1; string metric = 2; double value = 3; string unit = 4; }
message SummaryRequest  { string campaign_id = 1; }
message RecordResponse  { string metric_id = 1; bool accepted = 2; }
message IngestSummary   { int32 ingested_count = 1; int32 failed_count = 2; repeated string errors = 3; }
message AlertSubscription { string workspace_id = 1; AlertSeverity min_severity = 2; }

// ─── Enums ────────────────────────────────────────────────────────────────
enum Channel {
  CHANNEL_UNSPECIFIED = 0;
  CHANNEL_META        = 1;
  CHANNEL_GOOGLE      = 2;
  CHANNEL_TIKTOK      = 3;
}
enum AlertSeverity {
  SEVERITY_UNSPECIFIED = 0;
  SEVERITY_LOW         = 1;
  SEVERITY_MEDIUM      = 2;
  SEVERITY_HIGH        = 3;
  SEVERITY_CRITICAL    = 4;
}
enum CampaignStatus {
  STATUS_UNSPECIFIED = 0;
  STATUS_ACTIVE      = 1;
  STATUS_PAUSED      = 2;
  STATUS_COMPLETED   = 3;
}

const grpc        = require('@grpc/grpc-js');
const protoLoader = require('@grpc/proto-loader');
const crypto      = require('crypto');
const reflection  = require('@grpc/reflection');

// ── Carga dinámica del proto (sin compilar) ───────────────────────────────
const pkgDef = protoLoader.loadSync('proto/campaign_metrics.proto', {
  keepCase: true, longs: String, enums: String,
  defaults: true, oneofs: true,
});
const proto = grpc.loadPackageDefinition(pkgDef).dataflowai.metrics.v1;

// ── In-memory store (producción → PostgreSQL + Redis) ─────────────────────
const metricsStore  = new Map(); // campaign_id → [CampaignMetric]
const alertListeners = new Map(); // workspace_id → Set(call)

// ── Interceptor de logging ────────────────────────────────────────────────
function loggingInterceptor(methodDef, next) {
  return function(call, callback) {
    const start = Date.now();
    console.log(`→ [${methodDef.path}] llamada iniciada`);
    next(call, (...args) => {
      console.log(`← [${methodDef.path}] ${Date.now() - start}ms`);
      callback(...args);
    });
  };
}

// ── Helpers ───────────────────────────────────────────────────────────────
function computeSummary(campaignId) {
  const metrics = metricsStore.get(campaignId) || [];
  const totalSpend = metrics.reduce((s, m) => s + m.spend_eur, 0);
  const totalClicks = metrics.reduce((s, m) => s + m.clicks, 0);
  const totalImpr  = metrics.reduce((s, m) => s + m.impressions, 0);
  const totalConv  = metrics.reduce((s, m) => s + m.conversions, 0);
  const revenue    = totalConv * 47.5; // AOV estimado
  return {
    campaign_id:    campaignId,
    name:          `Campaña ${campaignId.slice(0,8)}`,
    total_spend_eur: totalSpend,
    roas:           totalSpend > 0 ? revenue / totalSpend : 0,
    ctr_pct:        totalImpr  > 0 ? (totalClicks / totalImpr) * 100 : 0,
    conversions:    totalConv,
    status:        'STATUS_ACTIVE',
  };
}

// ── Handlers ──────────────────────────────────────────────────────────────
const service = {

  // UNARY — guardar una métrica
  RecordMetric(call, callback) {
    const { campaign_id } = call.request;
    if (!campaign_id)
      return callback({ code: grpc.status.INVALID_ARGUMENT, message: 'campaign_id es obligatorio' });
    if (!metricsStore.has(campaign_id)) metricsStore.set(campaign_id, []);
    metricsStore.get(campaign_id).push(call.request);
    callback(null, { metric_id: crypto.randomUUID(), accepted: true });
  },

  // UNARY — resumen de campaña
  GetCampaignSummary(call, callback) {
    const { campaign_id } = call.request;
    if (!metricsStore.has(campaign_id))
      return callback({ code: grpc.status.NOT_FOUND, message: `Campaña ${campaign_id} no encontrada` });
    callback(null, computeSummary(campaign_id));
  },

  // CLIENT STREAMING — ingesta masiva
  BatchIngestMetrics(call, callback) {
    let ingested = 0; const errors = [];
    call.on('data', (metric) => {
      if (!metric.campaign_id) { errors.push('metric sin campaign_id omitida'); return; }
      if (!metricsStore.has(metric.campaign_id)) metricsStore.set(metric.campaign_id, []);
      metricsStore.get(metric.campaign_id).push(metric);
      ingested++;
    });
    call.on('end', () => callback(null, { ingested_count: ingested, failed_count: errors.length, errors }));
    call.on('error', (e) => console.error('BatchIngestMetrics error:', e));
  },

  // SERVER STREAMING — push alertas al dashboard
  StreamAlerts(call) {
    const { workspace_id, min_severity } = call.request;
    if (!alertListeners.has(workspace_id)) alertListeners.set(workspace_id, new Set());
    const listeners = alertListeners.get(workspace_id);
    listeners.add(call);
    // Simulación de detección de anomalía (producción → ML anomaly detector)
    const timer = setInterval(() => {
      call.write({
        campaign_id: 'camp_summer_2025', severity: 'SEVERITY_HIGH',
        message: 'CTR caído 40% en TikTok — posible fatiga creativa',
        metric_name: 'ctr_pct', value: 0.8, threshold: 1.5,
        fired_at: { seconds: Math.floor(Date.now() / 1000) },
      });
    }, 10000);
    call.on('cancelled', () => { clearInterval(timer); listeners.delete(call); });
  },

  // BIDIRECCIONAL — drill-down interactivo
  LiveMetricsChat(call) {
    call.on('data', ({ campaign_id, metric }) => {
      const summary = computeSummary(campaign_id);
      const value   = summary[metric] ?? 0;
      call.write({ campaign_id, metric, value, unit: unitOf(metric) });
    });
    call.on('end', () => call.end());
  },
};

// ── Arranque del servidor ─────────────────────────────────────────────────
const server = new grpc.Server({ interceptors: [loggingInterceptor] });
server.addService(proto.CampaignMetricsService.service, service);
reflection.addToServer(server); // habilita grpcurl en desarrollo
server.bindAsync('0.0.0.0:50055', grpc.ServerCredentials.createInsecure(), (err, port) => {
  if (err) { console.error('Error arrancando servidor:', err); return; }
  console.log(`✓ CampaignMetricsService escuchando en 0.0.0.0:${port}`);
});

"""DataFlow AI — Pipeline de ingesta masiva de métricas via gRPC client streaming."""
import grpc, csv, datetime
from google.protobuf import timestamp_pb2
import campaign_metrics_pb2 as pb
import campaign_metrics_pb2_grpc as stub

CHANNEL_MAP = { 'META': pb.CHANNEL_META, 'GOOGLE': pb.CHANNEL_GOOGLE, 'TIKTOK': pb.CHANNEL_TIKTOK }

def parse_csv_row(row):
    ts = timestamp_pb2.Timestamp()
    ts.FromDatetime(datetime.datetime.fromisoformat(row['timestamp']))
    return pb.CampaignMetric(
        campaign_id  = row['campaign_id'],
        channel      = CHANNEL_MAP.get(row['channel'], pb.CHANNEL_UNSPECIFIED),
        impressions  = int(row['impressions']),
        clicks       = int(row['clicks']),
        spend_eur   = float(row['spend_eur']),
        conversions = int(row['conversions']),
        timestamp    = ts,
    )

def metric_generator(csv_path):
    with open(csv_path) as f:
        for row in csv.DictReader(f):
            yield parse_csv_row(row)

def run_batch_ingest(csv_path: str):
    # Deadline de 60s para batch grandes
    with grpc.insecure_channel('localhost:50055') as channel:
        client   = stub.CampaignMetricsServiceStub(channel)
        response = client.BatchIngestMetrics(
            metric_generator(csv_path),
            timeout=60,
        )
        print(f"✓ Ingestadas: {response.ingested_count} | Fallidas: {response.failed_count}")
        if response.errors:
            print("Errores:", response.errors)

if __name__ == '__main__':
    run_batch_ingest('data/meta_export_2025-06-14.csv')
    # Ejemplo CSV esperado:
    # campaign_id,channel,impressions,clicks,spend_eur,conversions,timestamp
    # camp_summer_2025,META,150000,2100,840.50,38,2025-06-14T08:00:00
    # camp_summer_2025,TIKTOK,95000,760,310.00,12,2025-06-14T08:00:00
    # camp_brandawareness,GOOGLE,200000,4500,1200.00,55,2025-06-14T09:00:00

# Listar servicios disponibles (reflection activa)
grpcurl -plaintext localhost:50055 list
# → dataflowai.metrics.v1.CampaignMetricsService

# Inspeccionar RPC
grpcurl -plaintext localhost:50055 describe dataflowai.metrics.v1.CampaignMetricsService

# Registrar una métrica (unary)
grpcurl -plaintext -d '{
  "campaign_id": "camp_summer_2025",
  "channel": "CHANNEL_META",
  "impressions": 150000,
  "clicks": 2100,
  "spend_eur": 840.50,
  "conversions": 38
}' localhost:50055 dataflowai.metrics.v1.CampaignMetricsService/RecordMetric
# → { "metric_id": "3f7a1c2e-...", "accepted": true }

# Consultar resumen de campaña
grpcurl -plaintext -d '{"campaign_id": "camp_summer_2025"}' \
  localhost:50055 dataflowai.metrics.v1.CampaignMetricsService/GetCampaignSummary
# → { "campaign_id": "camp_summer_2025", "total_spend_eur": 840.5,
#     "roas": 2.15, "ctr_pct": 1.40, "conversions": 38, "status": "STATUS_ACTIVE" }

# Suscribirse a alertas en tiempo real (server streaming)
grpcurl -plaintext -d '{
  "workspace_id": "ws_dataflowai_001",
  "min_severity": "SEVERITY_MEDIUM"
}' localhost:50055 dataflowai.metrics.v1.CampaignMetricsService/StreamAlerts