Incremental Content Migration Scripts for Headless

import Database from 'better-sqlite3';
import { randomInt } from 'crypto';

// Configuration & Types
interface SyncState {
  content_id: string;
  legacy_updated_at: string;
  headless_etag: string | null;
  sync_status: 'pending' | 'success' | 'failed';
}

interface WebhookPayload {
  type: 'content_delta';
  affected_slugs: string[];
  version_hashes: Record<string, string>;
  timestamp: string;
}

interface CMSResponse {
  id: string;
  etag: string;
  slug: string;
}

class HeadlessDeltaSync {
  private db: Database.Database;
  private cmsBaseUrl: string;
  private cmsApiKey: string;
  private webhookUrl: string;
  private readonly BATCH_LIMIT = 500;
  private readonly MAX_RETRIES = 5;

  constructor(dbPath: string, cmsBaseUrl: string, cmsApiKey: string, webhookUrl: string) {
    this.db = new Database(dbPath);
    this.cmsBaseUrl = cmsBaseUrl;
    this.cmsApiKey = cmsApiKey;
    this.webhookUrl = webhookUrl;
    this.initializeStateTable();
  }

  private initializeStateTable(): void {
    this.db.exec(`
      CREATE TABLE IF NOT EXISTS sync_state (
        content_id TEXT PRIMARY KEY,
        legacy_updated_at TEXT NOT NULL,
        headless_etag TEXT,
        sync_status TEXT DEFAULT 'pending'
      );
      CREATE INDEX IF NOT EXISTS idx_updated_at ON sync_state(legacy_updated_at);
    `);
  }

  private getCursor(): string {
    const row = this.db.prepare('SELECT MAX(legacy_updated_at) as cursor FROM sync_state WHERE sync_status = ?').get('success') as { cursor: string | null };
    return row.cursor ?? '1970-01-01T00:00:00Z';
  }

  private async fetchLegacyDelta(cursor: string): Promise<Array<Record<string, any>>> {
    // Simulated legacy DB query. Replace with actual ORM/SQL driver.
    console.log(`[Delta] Fetching records updated after ${cursor}...`);
    return []; // Placeholder for actual legacy data fetch
  }

  private async upsertToCMS(record: Record<string, any>, etag: string | null): Promise<CMSResponse | null> {
    const endpoint = record.status === 'draft' 
      ? `${this.cmsBaseUrl}/api/v1/drafts` 
      : `${this.cmsBaseUrl}/api/v1/content`;

    const headers: Record<string, string> = {
      'Authorization': `Bearer ${this.cmsApiKey}`,
      'Content-Type': 'application/json',
      'Idempotency-Key': `${record.id}-${record.updated_at}`
    };

    if (etag) {
      // Enforce optimistic concurrency control per MDN specifications
      headers['If-Match'] = etag;
    }

    let attempt = 0;
    while (attempt < this.MAX_RETRIES) {
      try {
        const res = await fetch(endpoint, {
          method: 'POST',
          headers,
          body: JSON.stringify(record)
        });

        if (res.status === 429) {
          const retryAfter = parseInt(res.headers.get('Retry-After') || '5', 10);
          await new Promise(r => setTimeout(r, retryAfter * 1000 + randomInt(0, 1000)));
          attempt++;
          continue;
        }

        if (!res.ok) throw new Error(`CMS Error: ${res.status} ${res.statusText}`);

        const data = await res.json();
        return { id: data.id, etag: res.headers.get('ETag') || '', slug: data.slug };
      } catch (err) {
        attempt++;
        if (attempt >= this.MAX_RETRIES) throw err;
        const backoff = Math.pow(2, attempt) * 100 + randomInt(0, 500);
        await new Promise(r => setTimeout(r, backoff));
      }
    }
    return null;
  }

  private persistState(record: Record<string, any>, cmsResponse: CMSResponse | null): void {
    const stmt = this.db.prepare(`
      INSERT INTO sync_state (content_id, legacy_updated_at, headless_etag, sync_status)
      VALUES (?, ?, ?, ?)
      ON CONFLICT(content_id) DO UPDATE SET
        legacy_updated_at = excluded.legacy_updated_at,
        headless_etag = excluded.headless_etag,
        sync_status = excluded.sync_status
    `);
    stmt.run(
      record.id,
      record.updated_at,
      cmsResponse?.etag ?? null,
      cmsResponse ? 'success' : 'failed'
    );
  }

  private async dispatchTargetedWebhook(affectedRecords: CMSResponse[]): Promise<void> {
    if (affectedRecords.length === 0) return;

    const payload: WebhookPayload = {
      type: 'content_delta',
      affected_slugs: affectedRecords.map(r => r.slug),
      version_hashes: affectedRecords.reduce((acc, r) => ({ ...acc, [r.id]: r.etag }), {}),
      timestamp: new Date().toISOString()
    };

    await fetch(this.webhookUrl, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(payload)
    });
  }

  public async runSync(): Promise<void> {
    const cursor = this.getCursor();
    const deltaRecords = await this.fetchLegacyDelta(cursor);
    const successfulUpserts: CMSResponse[] = [];

    for (const record of deltaRecords) {
      const state = this.db.prepare('SELECT headless_etag FROM sync_state WHERE content_id = ?').get(record.id) as { headless_etag: string | null } | undefined;
      const currentEtag = state?.headless_etag ?? null;

      const cmsRes = await this.upsertToCMS(record, currentEtag);
      if (cmsRes) {
        this.persistState(record, cmsRes);
        successfulUpserts.push(cmsRes);
      }
    }

    await this.dispatchTargetedWebhook(successfulUpserts);
    console.log(`[Sync] Completed. Processed ${deltaRecords.length} records. Dispatched ${successfulUpserts.length} rebuild triggers.`);
  }
}

// Usage
// const sync = new HeadlessDeltaSync('./sync-state.db', 'https://cms.example.com', 'sk_live_...', 'https://frontend.example.com/api/revalidate');
// await sync.runSync();