Stream Analytics (PULSE)

// Track custom events for deeper analytics
import { WaveAnalytics } from '@wave/analytics';

const analytics = new WaveAnalytics({ apiKey: process.env.WAVE_API_KEY });

// Track viewer subscription
await analytics.track({
  event: 'subscription_started',
  userId: 'user_xyz789',
  streamId: 'stream_abc123',
  properties: {
    tier: 'premium',
    price: 9.99,
    currency: 'USD',
    trial: false,
    referrer: 'twitter_campaign'
  }
});

// Track purchase in stream
await analytics.track({
  event: 'purchase_completed',
  userId: 'user_xyz789',
  streamId: 'stream_abc123',
  properties: {
    product_id: 'merch_001',
    product_name: 'Team Jersey',
    price: 49.99,
    quantity: 1,
    category: 'merchandise'
  }
});

// Track engagement milestones
await analytics.track({
  event: 'engagement_milestone',
  userId: 'user_xyz789',
  streamId: 'stream_abc123',
  properties: {
    milestone: 'first_chat_message',
    watch_time_minutes: 5,
    session_number: 1
  }
});

// Track quality issues (auto-collected, but can supplement)
await analytics.track({
  event: 'quality_issue_reported',
  userId: 'user_xyz789',
  streamId: 'stream_abc123',
  properties: {
    issue_type: 'buffering',
    duration_seconds: 3,
    user_reported: true,
    network_type: '4g'
  }
});

// Configure analytics alerts
import { WaveAnalytics } from '@wave/analytics';

const analytics = new WaveAnalytics({ apiKey: process.env.WAVE_API_KEY });

// Viewer milestone alerts
await analytics.alerts.create({
  name: 'Viewer Milestone',
  type: 'threshold',
  metric: 'concurrent_viewers',
  condition: 'greater_than',
  thresholds: [100, 500, 1000, 5000, 10000],
  channels: ['webhook', 'email'],
  webhook_url: 'https://yourapp.com/webhooks/milestones',
  cooldown_minutes: 5
});

// Quality degradation alert
await analytics.alerts.create({
  name: 'Quality Alert',
  type: 'threshold',
  metric: 'rebuffer_rate',
  condition: 'greater_than',
  threshold: 5.0,  // 5% rebuffer rate
  channels: ['webhook', 'slack'],
  slack_channel: '#streaming-ops',
  severity: 'warning'
});

// Revenue target alert
await analytics.alerts.create({
  name: 'Revenue Goal',
  type: 'cumulative',
  metric: 'total_revenue',
  condition: 'greater_than',
  threshold: 10000,  // $10,000
  period: 'day',
  channels: ['email'],
  recipients: ['[email protected]']
});

// Anomaly detection (ML-powered)
await analytics.alerts.create({
  name: 'Traffic Anomaly',
  type: 'anomaly',
  metric: 'concurrent_viewers',
  sensitivity: 'medium',  // low, medium, high
  channels: ['webhook', 'pagerduty'],
  severity: 'critical'
});

// Export analytics to data warehouses
import { WaveAnalytics } from '@wave/analytics';

const analytics = new WaveAnalytics({ apiKey: process.env.WAVE_API_KEY });

// Configure BigQuery export
await analytics.exports.configure({
  destination: 'bigquery',
  project_id: 'your-gcp-project',
  dataset: 'wave_analytics',
  credentials: process.env.GCP_SERVICE_ACCOUNT_KEY,
  tables: {
    events: 'stream_events',
    metrics: 'stream_metrics',
    sessions: 'viewer_sessions'
  },
  schedule: 'hourly',  // hourly, daily, realtime
  partitioning: {
    field: 'timestamp',
    type: 'DAY'
  }
});

// Configure Snowflake export
await analytics.exports.configure({
  destination: 'snowflake',
  account: 'your-account.us-east-1',
  warehouse: 'WAVE_WH',
  database: 'ANALYTICS',
  schema: 'STREAMING',
  credentials: {
    user: process.env.SNOWFLAKE_USER,
    private_key: process.env.SNOWFLAKE_PRIVATE_KEY
  },
  schedule: 'daily'
});

// S3 export for custom processing
await analytics.exports.configure({
  destination: 's3',
  bucket: 'your-analytics-bucket',
  prefix: 'wave-analytics/',
  format: 'parquet',  // parquet, json, csv
  compression: 'snappy',
  partition_by: ['date', 'stream_id'],
  schedule: 'hourly'
});

// Webhook for real-time streaming
await analytics.exports.configure({
  destination: 'webhook',
  url: 'https://your-pipeline.com/ingest',
  format: 'json',
  batch_size: 100,
  batch_timeout_ms: 5000,
  headers: {
    'Authorization': 'Bearer YOUR_PIPELINE_TOKEN'
  }
});

// Create and analyze conversion funnels
import { WaveAnalytics } from '@wave/analytics';

const analytics = new WaveAnalytics({ apiKey: process.env.WAVE_API_KEY });

// Define a subscription conversion funnel
const funnel = await analytics.funnels.create({
  name: 'Subscription Conversion',
  steps: [
    { event: 'stream_viewed', label: 'Watched Stream' },
    { event: 'signup_started', label: 'Started Signup' },
    { event: 'signup_completed', label: 'Completed Signup' },
    { event: 'trial_started', label: 'Started Trial' },
    { event: 'subscription_started', label: 'Subscribed' }
  ],
  window_days: 7,  // 7-day conversion window
  breakdown_by: ['country', 'device', 'referrer']
});

// Get funnel analytics
const funnelData = await analytics.funnels.get(funnel.id, {
  start: new Date(Date.now() - 30 * 86400000),  // Last 30 days
  end: new Date(),
  breakdown: 'referrer'
});

console.log('Funnel Analysis:');
funnelData.steps.forEach((step, i) => {
  const conversionRate = i > 0
    ? ((step.count / funnelData.steps[i-1].count) * 100).toFixed(1)
    : '100.0';
  console.log(`${step.label}: ${step.count.toLocaleString()} (${conversionRate}%)`);
});

console.log('\nBreakdown by Referrer:');
funnelData.breakdown.forEach(segment => {
  console.log(`${segment.value}: ${segment.conversion_rate.toFixed(1)}% conversion`);
});