server/migrate_to_supabase.js

import { createClient } from '@supabase/supabase-js';
import Database from 'better-sqlite3';
import dotenv from 'dotenv';
import path from 'path';

dotenv.config();

const supabaseUrl = process.env.SUPABASE_URL;
const supabaseKey = process.env.SUPABASE_SERVICE_ROLE_KEY || process.env.SUPABASE_KEY;

if (!supabaseUrl || !supabaseKey || supabaseUrl.includes('YOUR_') || supabaseKey.includes('YOUR_')) {
  console.error('[Error] Supabase URL or Key is missing in .env!');
  process.exit(1);
}

console.log('[Migration] Connecting to SQLite...');
const sqliteDb = new Database('database.db');

console.log('[Migration] Connecting to Supabase...');
const supabase = createClient(supabaseUrl, supabaseKey);

// Fetch valid stream IDs
const streams = sqliteDb.prepare('SELECT id FROM streams').all();
const validStreamIds = new Set(streams.map(s => s.id));
console.log(`[Migration] Loaded ${validStreamIds.size} valid stream IDs from SQLite:`, Array.from(validStreamIds));

async function migrateStreams() {
  console.log('\n--- Migrating streams ---');
  const streamsData = sqliteDb.prepare('SELECT * FROM streams').all();
  if (streamsData.length === 0) {
    console.log('No streams to migrate.');
    return;
  }
  
  console.log(`Found ${streamsData.length} streams. Inserting...`);
  
  const { error } = await supabase
    .from('streams')
    .upsert(streamsData);
    
  if (error) {
    console.error('Error migrating streams:', error);
  } else {
    console.log('Successfully migrated streams!');
  }
}

async function migrateSettings() {
  console.log('\n--- Migrating settings ---');
  const settings = sqliteDb.prepare('SELECT * FROM settings').all();
  if (settings.length === 0) {
    console.log('No settings to migrate.');
    return;
  }
  
  console.log(`Found ${settings.length} settings. Inserting...`);
  const { error } = await supabase
    .from('settings')
    .upsert(settings);
    
  if (error) {
    console.error('Error migrating settings:', error);
  } else {
    console.log('Successfully migrated settings!');
  }
}

async function migrateModActions() {
  console.log('\n--- Migrating mod_actions ---');
  // Only migrate mod actions that reference valid stream IDs or have null stream_id
  const modActions = sqliteDb.prepare('SELECT * FROM mod_actions').all();
  if (modActions.length === 0) {
    console.log('No mod actions to migrate.');
    return;
  }
  
  const filteredActions = modActions.filter(act => act.stream_id === null || validStreamIds.has(act.stream_id));
  console.log(`Found ${modActions.length} mod actions. Migrating ${filteredActions.length} actions that have valid stream_id...`);
  
  if (filteredActions.length === 0) return;

  const { error } = await supabase
    .from('mod_actions')
    .upsert(filteredActions);
    
  if (error) {
    console.error('Error migrating mod actions:', error);
  } else {
    console.log('Successfully migrated mod actions!');
  }
}

async function migrateMessages() {
  console.log('\n--- Migrating messages ---');
  const total = sqliteDb.prepare('SELECT count(*) as c FROM messages').get().c;
  if (total === 0) {
    console.log('No messages to migrate.');
    return;
  }
  
  console.log(`Found ${total} messages to migrate.`);
  const batchSize = 1000;
  let migratedCount = 0;
  
  for (let offset = 0; offset < total; offset += batchSize) {
    const batch = sqliteDb.prepare('SELECT * FROM messages LIMIT ? OFFSET ?').all(batchSize, offset);
    
    // Filter messages that have valid stream_id
    const filteredBatch = batch.filter(msg => msg.stream_id === null || validStreamIds.has(msg.stream_id));
    
    if (filteredBatch.length === 0) continue;

    // Map SQLite types to Postgres types (booleans)
    const mappedBatch = filteredBatch.map(msg => ({
      id: msg.id,
      stream_id: msg.stream_id,
      username: msg.username,
      display_name: msg.display_name,
      message: msg.message,
      timestamp: msg.timestamp,
      is_streamer: !!msg.is_streamer,
      is_mod: !!msg.is_mod,
      is_sub: !!msg.is_sub
    }));
    
    const { error } = await supabase
      .from('messages')
      .upsert(mappedBatch);
      
    if (error) {
      console.error(`Error migrating messages batch at offset ${offset}:`, error);
      console.log('Retrying...');
      await new Promise(r => setTimeout(r, 1000));
      const { error: retryError } = await supabase.from('messages').upsert(mappedBatch);
      if (retryError) {
        console.error('Retry failed, skipping batch:', retryError);
      } else {
        migratedCount += mappedBatch.length;
      }
    } else {
      migratedCount += mappedBatch.length;
    }
    
    if (offset % 5000 === 0 || offset + batchSize >= total) {
      console.log(`[Messages Progress] Migrated ${Math.min(offset + batchSize, total)} / ${total} messages`);
    }
  }
  
  console.log(`Successfully migrated messages (inserted ${migratedCount} rows)!`);
}

async function migrateVoiceWords() {
  console.log('\n--- Migrating voice_words ---');
  const total = sqliteDb.prepare('SELECT count(*) as c FROM voice_words').get().c;
  if (total === 0) {
    console.log('No voice words to migrate.');
    return;
  }
  
  console.log(`Found ${total} voice words to migrate.`);
  const batchSize = 3000;
  let migratedCount = 0;
  
  for (let offset = 0; offset < total; offset += batchSize) {
    const batch = sqliteDb.prepare('SELECT * FROM voice_words LIMIT ? OFFSET ?').all(batchSize, offset);
    
    // Filter voice words that have valid stream_id
    const filteredBatch = batch.filter(vw => vw.stream_id === null || validStreamIds.has(vw.stream_id));
    
    if (filteredBatch.length === 0) continue;

    const { error } = await supabase
      .from('voice_words')
      .upsert(filteredBatch);
      
    if (error) {
      console.error(`Error migrating voice words batch at offset ${offset}:`, error);
      console.log('Retrying...');
      await new Promise(r => setTimeout(r, 1000));
      const { error: retryError } = await supabase.from('voice_words').upsert(filteredBatch);
      if (retryError) {
        console.error('Retry failed, skipping batch:', retryError);
      } else {
        migratedCount += filteredBatch.length;
      }
    } else {
      migratedCount += filteredBatch.length;
    }
    
    if (offset % 30000 === 0 || offset + batchSize >= total) {
      console.log(`[Voice Words Progress] Migrated ${Math.min(offset + batchSize, total)} / ${total} voice words`);
    }
  }
  
  console.log(`Successfully migrated voice words (inserted ${migratedCount} rows)!`);
}

async function startMigration() {
  console.log('=== Starting database migration from SQLite to Supabase ===');
  const startTime = Date.now();
  
  try {
    await migrateStreams();
    await migrateSettings();
    await migrateModActions();
    await migrateMessages();
    await migrateVoiceWords();
    
    const duration = ((Date.now() - startTime) / 1000).toFixed(1);
    console.log(`\n=== Migration completed successfully in ${duration}s! ===`);
    console.log('\nIMPORTANT: Now run the following SQL query in your Supabase SQL Editor to reset auto-increment sequences:');
    console.log(`
SELECT setval(pg_get_serial_sequence('streams', 'id'), coalesce(max(id), 1)) FROM streams;
SELECT setval(pg_get_serial_sequence('voice_words', 'id'), coalesce(max(id), 1)) FROM voice_words;
SELECT setval(pg_get_serial_sequence('mod_actions', 'id'), coalesce(max(id), 1)) FROM mod_actions;
    `);
  } catch (err) {
    console.error('Migration failed:', err);
  } finally {
    sqliteDb.close();
  }
}

startMigration();