sync.mjs

#!/usr/bin/env node

/**
 * Local Dev Sync Script - Bidirectional R2 ↔ Storage Sync
 *
 * R2 → Storage:
 * 1. Watches .wrangler/state/v3/r2/ for R2 changes
 * 2. Parses miniflare SQLite to get object keys and blob IDs
 * 3. Copies blobs to ./.storage/ with proper paths
 *
 * Storage → R2:
 * 4. Watches .storage/ for local file changes
 * 5. Uses `wrangler r2 object put/delete --local` to sync to R2
 *
 * Container Sync:
 * 6. Watches for sandbox container and runs mutagen sync
 */

import { watch } from 'chokidar';
import { execSync, spawn } from 'child_process';
import { existsSync, mkdirSync, copyFileSync, readdirSync, readFileSync } from 'fs';
import { dirname, join, relative } from 'path';
import Database from 'better-sqlite3';

// Paths (relative to monorepo root)
const MONOREPO_ROOT = join(import.meta.dirname, '../..');
const WRANGLER_R2_STATE = join(MONOREPO_ROOT, 'apps/sandbox/.wrangler/state/v3/r2');
const STORAGE_DIR = join(MONOREPO_ROOT, '.storage');

// R2 bucket name from wrangler.jsonc
const R2_BUCKET = 'builder-storage';

// Container pattern for mutagen
const CONTAINER_PATTERN = 'workerd-builder-sandbox';

// Wrangler CWD for R2 commands
const WRANGLER_CWD = join(MONOREPO_ROOT, 'apps/sandbox');

// State
let currentContainer = null;
let syncDebounceTimer = null;
let syncToR2Timer = null;
const SYNC_DEBOUNCE_MS = 500;
const SYNC_COOLDOWN_MS = 1000;

// Track recently synced keys to prevent sync loops
const recentR2Syncs = new Set();
const pendingSyncsToR2 = new Map();

console.log('========================================');
console.log('  Local Dev Sync (Bidirectional)');
console.log('========================================');
console.log(`  R2 State: ${WRANGLER_R2_STATE}`);
console.log(`  Storage:  ${STORAGE_DIR}`);
console.log(`  Wrangler: ${WRANGLER_CWD}`);
console.log('========================================\n');

// Ensure storage directory exists
if (!existsSync(STORAGE_DIR)) {
    mkdirSync(STORAGE_DIR, { recursive: true });
    console.log('📁 Created storage directory');
}

/**
 * Find the SQLite database file in miniflare-R2BucketObject/
 */
function findSqliteDb() {
    const dbDir = join(WRANGLER_R2_STATE, 'miniflare-R2BucketObject');
    if (!existsSync(dbDir)) {
        return null;
    }

    const files = readdirSync(dbDir).filter(f => f.endsWith('.sqlite'));
    if (files.length === 0) {
        return null;
    }

    // Return the first .sqlite file (there should only be one per bucket)
    return join(dbDir, files[0]);
}

/**
 * Sync R2 objects to ./storage/ by parsing miniflare SQLite
 */
async function syncR2ToStorage() {
    console.log('🔄 Syncing R2 to storage...');

    try {
        const dbPath = findSqliteDb();
        if (!dbPath) {
            console.log('   No SQLite database found. Run wrangler dev and write to R2 first.');
            return;
        }

        // Open SQLite database (read-only)
        const db = new Database(dbPath, { readonly: true });

        // Query all objects
        const objects = db.prepare('SELECT key, blob_id FROM _mf_objects').all();
        db.close();

        if (objects.length === 0) {
            console.log('   No objects in R2');
            return;
        }

        console.log(`   Found ${objects.length} objects`);

        // Blobs directory
        const blobsDir = join(WRANGLER_R2_STATE, R2_BUCKET, 'blobs');

        // Copy each blob to storage with proper path
        for (const obj of objects) {
            const { key, blob_id } = obj;
            const blobPath = join(blobsDir, blob_id);
            const storagePath = join(STORAGE_DIR, key);
            const storageDir = dirname(storagePath);

            // Ensure directory exists
            if (!existsSync(storageDir)) {
                mkdirSync(storageDir, { recursive: true });
            }

            // Copy blob to storage
            if (existsSync(blobPath)) {
                copyFileSync(blobPath, storagePath);
                // Mark as recently synced to prevent loop
                recentR2Syncs.add(key);
                setTimeout(() => recentR2Syncs.delete(key), SYNC_COOLDOWN_MS);
                console.log(`   ✅ ${key}`);
            } else {
                console.log(`   ⚠️  ${key}: blob not found`);
            }
        }

        console.log('🔄 R2 sync complete\n');
    } catch (err) {
        console.error('❌ R2 sync failed:', err.message);
    }
}

/**
 * Debounced sync - waits for changes to settle (R2 → Storage)
 */
function debouncedSync() {
    if (syncDebounceTimer) {
        clearTimeout(syncDebounceTimer);
    }
    syncDebounceTimer = setTimeout(() => {
        syncR2ToStorage();
    }, SYNC_DEBOUNCE_MS);
}

/**
 * Sync a single file from .storage/ to local R2 using wrangler CLI
 */
async function syncFileToR2(filePath, event) {
    const key = relative(STORAGE_DIR, filePath);
    const objectPath = `${R2_BUCKET}/${key}`;

    try {
        if (event === 'unlink') {
            // Delete from local R2
            execSync(`npx wrangler r2 object delete "${objectPath}" --local`, {
                cwd: WRANGLER_CWD,
                encoding: 'utf-8',
                stdio: 'pipe'
            });
            console.log(`   🗑️  Deleted from R2: ${key}`);
        } else {
            // Put file to local R2
            execSync(`npx wrangler r2 object put "${objectPath}" --file="${filePath}" --local`, {
                cwd: WRANGLER_CWD,
                encoding: 'utf-8',
                stdio: 'pipe'
            });
            console.log(`   ✅ Synced to R2: ${key}`);
        }
    } catch (err) {
        console.error(`   ❌ R2 sync failed for ${key}:`, err.message);
    }
}

/**
 * Debounced sync from .storage/ to R2
 */
function debouncedSyncToR2(filePath, event) {
    const key = relative(STORAGE_DIR, filePath);

    // Skip if recently synced FROM R2 (prevent loops)
    if (recentR2Syncs.has(key)) {
        return;
    }

    pendingSyncsToR2.set(filePath, event);

    if (syncToR2Timer) {
        clearTimeout(syncToR2Timer);
    }
    syncToR2Timer = setTimeout(async () => {
        if (pendingSyncsToR2.size === 0) return;

        console.log('🔄 Syncing storage to R2...');
        for (const [path, evt] of pendingSyncsToR2) {
            await syncFileToR2(path, evt);
        }
        pendingSyncsToR2.clear();
        console.log('🔄 Storage→R2 sync complete\n');
    }, SYNC_DEBOUNCE_MS);
}

/**
 * Find sandbox container
 */
function findContainer() {
    try {
        const result = execSync(
            `docker ps -qf "name=${CONTAINER_PATTERN}" | head -1`,
            { encoding: 'utf-8', timeout: 5000 }
        ).trim();
        return result || null;
    } catch {
        return null;
    }
}

/**
 * Start mutagen sync to container
 */
function startMutagenSync(containerId) {
    console.log(`📦 Starting mutagen sync to container ${containerId.slice(0, 12)}...`);

    try {
        // Terminate existing sync if any
        try {
            execSync('mutagen sync terminate storefront-storage', {
                encoding: 'utf-8',
                stdio: 'pipe'
            });
        } catch {
            // Ignore if doesn't exist
        }

        // Create new sync
        execSync(
            `mutagen sync create --name=storefront-storage "${STORAGE_DIR}" "docker://${containerId}/mnt/storage" --sync-mode=two-way-resolved --ignore-vcs --ignore=".DS_Store"`,
            { encoding: 'utf-8', timeout: 30000 }
        );

        // Wait for initial sync to complete before continuing
        console.log('⏳ Waiting for initial sync...');
        execSync('mutagen sync flush storefront-storage', {
            encoding: 'utf-8',
            timeout: 60000
        });

        console.log('✅ Mutagen sync active\n');
    } catch (err) {
        console.error('❌ Mutagen sync failed:', err.message);
    }
}

/**
 * Stop mutagen sync
 */
function stopMutagenSync() {
    try {
        execSync('mutagen sync terminate storefront-storage', {
            encoding: 'utf-8',
            stdio: 'pipe'
        });
        console.log('⏸️  Mutagen sync stopped\n');
    } catch {
        // Ignore
    }
}

/**
 * Check container status and manage mutagen
 */
function checkContainer() {
    const container = findContainer();

    if (container && container !== currentContainer) {
        // New container or container changed
        if (currentContainer) {
            console.log('🔄 Container changed, restarting mutagen...');
            stopMutagenSync();
        }
        currentContainer = container;
        startMutagenSync(container);
    } else if (!container && currentContainer) {
        // Container stopped
        console.log('⏸️  Container stopped');
        stopMutagenSync();
        currentContainer = null;
        console.log('🔄 Waiting for container...\n');
    }
}

/**
 * Watch for container events using docker events (real-time, no polling delay)
 */
function watchContainerEvents() {
    console.log('👁️  Watching docker events for container changes...');

    // Use docker events to watch for container start/die events
    const dockerEvents = spawn('docker', [
        'events',
        '--filter', 'type=container',
        '--filter', `name=${CONTAINER_PATTERN}`,
        '--format', '{{.Status}} {{.Actor.ID}}'
    ], { stdio: ['ignore', 'pipe', 'pipe'] });

    dockerEvents.stdout.on('data', (data) => {
        const lines = data.toString().trim().split('\n');
        for (const line of lines) {
            const [status, containerId] = line.split(' ');

            if (status === 'start' && containerId) {
                console.log(`🐳 Container started: ${containerId.slice(0, 12)}`);
                // Small delay to ensure container is fully ready
                setTimeout(() => {
                    if (containerId !== currentContainer) {
                        if (currentContainer) {
                            stopMutagenSync();
                        }
                        currentContainer = containerId;
                        startMutagenSync(containerId);
                    }
                }, 100);
            } else if ((status === 'die' || status === 'stop') && containerId === currentContainer) {
                console.log(`⏸️  Container stopped: ${containerId.slice(0, 12)}`);
                stopMutagenSync();
                currentContainer = null;
            }
        }
    });

    dockerEvents.stderr.on('data', (data) => {
        console.error('Docker events error:', data.toString());
    });

    dockerEvents.on('close', (code) => {
        console.log(`Docker events watcher closed with code ${code}, restarting...`);
        // Restart watcher if it closes unexpectedly
        setTimeout(watchContainerEvents, 1000);
    });

    return dockerEvents;
}

/**
 * Cleanup on exit
 */
let dockerEventsProcess = null;
function cleanup() {
    console.log('\n🛑 Shutting down...');
    if (dockerEventsProcess) {
        dockerEventsProcess.kill();
    }
    stopMutagenSync();
    process.exit(0);
}

process.on('SIGINT', cleanup);
process.on('SIGTERM', cleanup);

// ========================================
// Main
// ========================================

// Initial R2 sync
console.log('🚀 Starting initial R2 sync...\n');
await syncR2ToStorage();

// Watch R2 blobs directory for changes (not SQLite - reading it triggers WAL updates)
const blobsDir = join(WRANGLER_R2_STATE, R2_BUCKET, 'blobs');
let r2Watcher = null;

function startBlobsWatcher() {
    if (r2Watcher) return; // Already watching

    if (existsSync(blobsDir)) {
        console.log('👁️  Watching R2 blobs for changes...');

        r2Watcher = watch(blobsDir, {
            persistent: true,
            ignoreInitial: true,
            awaitWriteFinish: {
                stabilityThreshold: 300,
                pollInterval: 100
            }
        });

        r2Watcher.on('all', (event, filePath) => {
            console.log(`📝 R2 blob change: ${event} - ${filePath.split('/').pop()}`);
            debouncedSync();
        });
    }
}

// Try to start watcher now
startBlobsWatcher();

if (!r2Watcher) {
    console.log('👁️  Waiting for R2 blobs directory to be created...');
    // Poll for blobs directory to appear
    const blobsDirInterval = setInterval(() => {
        if (existsSync(blobsDir)) {
            clearInterval(blobsDirInterval);
            startBlobsWatcher();
            // Do initial sync when directory appears
            syncR2ToStorage();
        }
    }, 2000);
}

// Watch for container using docker events (real-time detection)
checkContainer(); // Initial check for existing container
dockerEventsProcess = watchContainerEvents();

// ========================================
// Storage → R2 Sync (reverse direction)
// ========================================

console.log('👁️  Watching .storage/ for local changes...');
const storageWatcher = watch(STORAGE_DIR, {
    persistent: true,
    ignoreInitial: true,
    ignored: ['.DS_Store', '.sync-state.json', /^\./],
    awaitWriteFinish: {
        stabilityThreshold: 300,
        pollInterval: 100
    }
});

storageWatcher.on('add', (path) => debouncedSyncToR2(path, 'add'));
storageWatcher.on('change', (path) => debouncedSyncToR2(path, 'change'));
storageWatcher.on('unlink', (path) => debouncedSyncToR2(path, 'unlink'));

// Keep process running
console.log('Press Ctrl+C to stop\n');