NutriTrack — Pipeline MotherDuck + DuckDB

# Autenticación MotherDuck — NutriTrack
import duckdb
import os

# Token en variable de entorno (jamás en código)
conn = duckdb.connect(
    "md:nutritrack_analytics",
    config={
        "motherduck_token": os.environ["MOTHERDUCK_TOKEN"]
    }
)

# Crear base de datos cloud (persiste entre sesiones)
conn.sql("CREATE DATABASE IF NOT EXISTS nutritrack")
conn.sql("USE nutritrack")

print("✅ Conectado a MotherDuck — nutritrack")

# Carga incremental desde S3 → MotherDuck
conn.sql("""
  CREATE TABLE IF NOT EXISTS events (
    event_id    VARCHAR,
    user_id     VARCHAR,
    event_type  VARCHAR,
    amount      DOUBLE,
    plan        VARCHAR,
    created_at  TIMESTAMP
  )
""")

# Append incremental — evita recargar 3 TB
conn.sql("""
  INSERT INTO events
  SELECT * FROM read_parquet(
    's3://nutritrack-data/events/2026-06-*.parquet',
    hive_partitioning = true
  )
  WHERE event_id NOT IN (SELECT event_id FROM events)
    AND created_at::DATE = current_date - 1
""")

print(f"Filas cargadas: {conn.sql('SELECT COUNT(*) FROM events').fetchone()[0]:,}")

# Tabla materializada de KPIs diarios
conn.sql("""
  CREATE OR REPLACE TABLE daily_kpis AS
  SELECT
    DATE(created_at)                                 AS date,
    COUNT(DISTINCT user_id)                          AS dau,
    COUNT(DISTINCT CASE WHEN event_type = 'signup'
      THEN user_id END)                              AS signups,
    COUNT(DISTINCT CASE WHEN event_type = 'churn'
      THEN user_id END)                              AS churns,
    COUNT(DISTINCT CASE WHEN event_type = 'premium_upgrade'
      THEN user_id END)                              AS upgrades,
    SUM(CASE WHEN event_type = 'premium_upgrade'
      THEN amount ELSE 0 END)                        AS revenue,
    COUNT(*)                                         AS total_events
  FROM events
  WHERE created_at >= current_date - INTERVAL 90 DAYS
  GROUP BY 1
  ORDER BY 1 DESC
""")
print("✅ daily_kpis actualizada")

// API de analytics — expone MotherDuck a frontend
import duckdb from "duckdb-async"

const db = await duckdb.Database.create(
  "md:nutritrack",
  { motherduck_token: process.env.MOTHERDUCK_TOKEN! }
)

async function getKPIs(days: number = 30) {
  return db.all(`
    SELECT date, dau, signups, revenue, churns
    FROM daily_kpis
    WHERE date >= current_date - INTERVAL '${days} days'
    ORDER BY date
  `)
}

// Endpoint Express
app.get("/api/kpis", async (req, res) => {
  const data = await getKPIs(parseInt(req.query.days) || 30)
  res.json({ ok: true, data })
})