Phase 4: Proactive Intelligence

const WatchUrlSchema = Type.Object({
  url: Type.String({ format: 'uri' }),
  url_type: Type.Optional(Type.Union([
    Type.Literal('page'),
    Type.Literal('sitemap'),
    Type.Literal('blog'),
    Type.Literal('news')
  ])),
  check_frequency_hours: Type.Optional(Type.Number({ minimum: 1, maximum: 168 }))
});

app.post<{
  Params: { id: string },
  Body: typeof WatchUrlSchema.static
}>('/client/:id/watch_url', {
  schema: {
    body: WatchUrlSchema
  }
}, async (request, reply) => {
  const { id } = request.params;
  const { url, url_type = 'page', check_frequency_hours = 6 } = request.body;

  // Fetch initial content and hash
  const { markdown } = await scrapeUrl(url);
  const hash = crypto.createHash('sha256').update(markdown).digest('hex');

  const { data, error } = await supabase
    .from('watched_urls')
    .upsert({
      customer_id: parseInt(id),
      url,
      url_type,
      last_hash: hash,
      last_checked_at: new Date().toISOString(),
      check_frequency_hours,
      is_active: true
    }, {
      onConflict: 'customer_id,url'
    })
    .select()
    .single();

  if (error) {
    return reply.code(500).send({ error: error.message });
  }

  return data;
});

import crypto from 'crypto';
import { scrapeUrl } from './scraper.js';
import { createSupabaseClient } from './supabase.js';

const supabase = createSupabaseClient();

export async function checkUrlForChanges(watchedUrl: any): Promise<boolean> {
  try {
    const { markdown } = await scrapeUrl(watchedUrl.url);
    const newHash = crypto.createHash('sha256').update(markdown).digest('hex');

    if (newHash !== watchedUrl.last_hash) {
      // Content changed!
      await supabase
        .from('watched_urls')
        .update({
          last_hash: newHash,
          last_changed_at: new Date().toISOString(),
          last_checked_at: new Date().toISOString()
        })
        .eq('id', watchedUrl.id);

      return true; // Changed
    } else {
      // No change
      await supabase
        .from('watched_urls')
        .update({ last_checked_at: new Date().toISOString() })
        .eq('id', watchedUrl.id);

      return false;
    }
  } catch (error) {
    console.error(`Error checking URL ${watchedUrl.url}:`, error);
    return false;
  }
}

export async function checkAllWatchedUrls() {
  const now = new Date();

  // Get URLs that need checking
  const { data: urls, error } = await supabase
    .from('watched_urls')
    .select('*')
    .eq('is_active', true)
    .or(`last_checked_at.is.null,last_checked_at.lt.${new Date(now.getTime() - 6 * 60 * 60 * 1000).toISOString()}`);

  if (error || !urls) {
    console.error('Error fetching watched URLs:', error);
    return;
  }

  const changedUrls: any[] = [];

  for (const url of urls) {
    const changed = await checkUrlForChanges(url);
    if (changed) {
      changedUrls.push(url);
    }
  }

  return changedUrls;
}

import cron from 'node-cron';
import { checkAllWatchedUrls } from '../utils/url-monitor.js';
import { env } from '../config/env.js';

export function startUrlMonitoring() {
  // Run every hour
  cron.schedule('0 * * * *', async () => {
    console.log('Running URL monitoring check...');
    
    const changedUrls = await checkAllWatchedUrls();
    
    if (changedUrls && changedUrls.length > 0) {
      console.log(`Found ${changedUrls.length} changed URLs`);
      
      // Trigger GoClaw agent for each change
      for (const url of changedUrls) {
        await triggerContentGeneration(url);
      }
    }
  });
}

async function triggerContentGeneration(watchedUrl: any) {
  // Call GoClaw API to trigger creator-agent
  const goclawUrl = `${env.GOCLAW_API_URL}/sessions/message`;
  
  // Get telegram_chat_id for this customer
  const { data: mapping } = await supabase
    .from('telegram_client_mapping')
    .select('telegram_chat_id')
    .eq('customer_id', watchedUrl.customer_id)
    .single();

  if (!mapping) return;

  await fetch(goclawUrl, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      channel: 'telegram',
      chat_id: mapping.telegram_chat_id,
      message: `I noticed your website was updated: ${watchedUrl.url}. Would you like me to create content about the changes?`,
      agent_id: 'liaison-agent'
    })
  });
}

import { startUrlMonitoring } from './scheduler/url-monitor.js';

startUrlMonitoring();

app.get('/watched-urls/check', async (request, reply) => {
  const changedUrls = await checkAllWatchedUrls();
  return changedUrls || [];
});

import { XMLParser } from 'fast-xml-parser';

export async function parseSitemap(sitemapUrl: string): Promise<string[]> {
  const response = await fetch(sitemapUrl);
  const xml = await response.text();

  const parser = new XMLParser();
  const parsed = parser.parse(xml);

  const urls: string[] = [];

  // Handle standard sitemap
  if (parsed.urlset?.url) {
    const urlEntries = Array.isArray(parsed.urlset.url)
      ? parsed.urlset.url
      : [parsed.urlset.url];

    for (const entry of urlEntries) {
      urls.push(entry.loc);
    }
  }

  // Handle sitemap index
  if (parsed.sitemapindex?.sitemap) {
    const sitemaps = Array.isArray(parsed.sitemapindex.sitemap)
      ? parsed.sitemapindex.sitemap
      : [parsed.sitemapindex.sitemap];

    for (const sm of sitemaps) {
      const subUrls = await parseSitemap(sm.loc);
      urls.push(...subUrls);
    }
  }

  return urls;
}

app.post<{ Params: { id: string }, Body: { sitemap_url: string } }>(
  '/client/:id/watch_sitemap',
  async (request, reply) => {
    const { id } = request.params;
    const { sitemap_url } = request.body;

    const urls = await parseSitemap(sitemap_url);

    // Watch all URLs found in sitemap
    const results = [];
    for (const url of urls) {
      const { data } = await supabase
        .from('watched_urls')
        .upsert({
          customer_id: parseInt(id),
          url,
          url_type: 'blog',
          is_active: true,
          check_frequency_hours: 24
        }, {
          onConflict: 'customer_id,url'
        })
        .select();

      results.push(data);
    }

    return {
      discovered: urls.length,
      watching: results.length
    };
  }
);

const UpcomingEventsQuerySchema = Type.Object({
  industries: Type.Optional(Type.Array(Type.String())),
  days: Type.Optional(Type.Number({ minimum: 1, maximum: 90 }))
});

app.get<{ Querystring: typeof UpcomingEventsQuerySchema.static }>(
  '/calendar/upcoming',
  {
    schema: {
      querystring: UpcomingEventsQuerySchema
    }
  },
  async (request, reply) => {
    const { industries = [], days = 14 } = request.query;

    const endDate = new Date();
    endDate.setDate(endDate.getDate() + days);

    let query = supabase
      .from('marketing_calendar')
      .select('*')
      .gte('event_date', new Date().toISOString().split('T')[0])
      .lte('event_date', endDate.toISOString().split('T')[0])
      .order('event_date', { ascending: true });

    // Filter by industries if provided
    if (industries.length > 0) {
      query = query.or(
        `industries.cs.{${industries.join(',')}},is_global.eq.true`
      );
    }

    const { data, error } = await query;

    if (error) {
      return reply.code(500).send({ error: error.message });
    }

    return data || [];
  }
);

import { TwitterApi } from 'twitter-api-v2';

export async function fetchTwitterAnalytics(
  client: TwitterApi,
  tweetId: string
): Promise<{
  likes: number;
  retweets: number;
  replies: number;
  impressions: number;
}> {
  const tweet = await client.v2.singleTweet(tweetId, {
    'tweet.fields': ['public_metrics']
  });

  return {
    likes: tweet.data.public_metrics?.like_count || 0,
    retweets: tweet.data.public_metrics?.retweet_count || 0,
    replies: tweet.data.public_metrics?.reply_count || 0,
    impressions: tweet.data.public_metrics?.impression_count || 0
  };
}

import mailchimp from '@mailchimp/mailchimp_transactional';

export async function fetchEmailAnalytics(
  apiKey: string,
  messageId: string
): Promise<{
  opens: number;
  clicks: number;
  open_rate: number;
  click_rate: number;
}> {
  const client = mailchimp(apiKey);
  
  const info = await client.messages.info({ id: messageId });
  
  return {
    opens: info.opens,
    clicks: info.clicks,
    open_rate: info.opens / Math.max(info.sends, 1),
    click_rate: info.clicks / Math.max(info.opens, 1)
  };
}

app.post<{ Params: { id: string } }>(
  '/client/:id/analytics/aggregate',
  async (request, reply) => {
    const { id } = request.params;

    // Fetch all published posts from last 30 days with analytics
    const { data: posts, error } = await supabase
      .from('customer_platform_post')
      .select(`
        id,
        platform,
        content,
        customer_post:customer_posts!inner(customer_id),
        analytics:post_analytics(metric_type, metric_value)
      `)
      .eq('customer_post.customer_id', parseInt(id))
      .gte('customer_post.created_at', new Date(Date.now() - 30 * 24 * 60 * 60 * 1000).toISOString());

    if (error || !posts) {
      return reply.code(500).send({ error: error?.message || 'No data' });
    }

    // Analyze patterns
    const insights = analyzePostPerformance(posts);

    return insights;
  }
);

function analyzePostPerformance(posts: any[]): Record<string, any> {
  // Group by platform
  const byPlatform = posts.reduce((acc, post) => {
    if (!acc[post.platform]) acc[post.platform] = [];
    acc[post.platform].push(post);
    return acc;
  }, {} as Record<string, any[]>);

  const insights: Record<string, any> = {};

  for (const [platform, platformPosts] of Object.entries(byPlatform)) {
    // Calculate averages
    const avgLikes = platformPosts.reduce((sum, p) =>
      sum + (p.analytics.find((a: any) => a.metric_type === 'likes')?.metric_value || 0), 0
    ) / platformPosts.length;

    // Find top performer
    const topPost = platformPosts.sort((a, b) => {
      const aLikes = a.analytics.find((m: any) => m.metric_type === 'likes')?.metric_value || 0;
      const bLikes = b.analytics.find((m: any) => m.metric_type === 'likes')?.metric_value || 0;
      return bLikes - aLikes;
    })[0];

    insights[platform] = {
      post_count: platformPosts.length,
      avg_likes: avgLikes,
      top_post_content: topPost?.content.substring(0, 100),
      top_post_performance: topPost?.analytics
    };
  }

  return insights;
}

// Before creating new post, check unsent count
const { count, error: countError } = await supabase
  .from('customer_posts')
  .select('*', { count: 'exact', head: true })
  .eq('customer_id', customer_id)
  .in('status', ['new', 'pending_review', 'approved']);

const preferences = customerData.scheduling_preferences || { max_unsent_posts: 3 };

if (count && count >= preferences.max_unsent_posts) {
  return reply.code(429).send({
    error: `Maximum of ${preferences.max_unsent_posts} unsent posts reached. Please approve or reject existing posts first.`
  });
}