refactor: remove DeltaLake dependency, replace with ClickHouse pipeline

DeltaLake was never fully wired up (factory files never created).
Now that events flow through Pipeline → R2 → S3Queue → ClickHouse,
remove the dead DeltaLake code paths:

- Remove writeEventsDelta/writeEventsDualWrite from event-writer.ts
- Remove USE_DELTALAKE/USE_DELTALAKE_CDC/DELTALAKE_DUAL_WRITE env vars
- Remove DeltaLake compaction tasks from scheduled handler
- Simplify EventWriterDO flush to legacy Parquet only
- Add missing isCollectionChangeEvent export to cdc-processor.ts
- Add headlessly-tail to tail_consumers

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: nathanclevenger

core/src/cdc-processor.ts

@@ -23,6 +23,13 @@ import {
 export type { CDCEvent } from './cdc-delta.js'
 export { isCdcEvent } from './types.js'
 
+/**
+ * Type guard for collection change events (CDC: insert/update/delete).
+ */
+export function isCollectionChangeEvent(event: { type?: string }): boolean {
+  return typeof event.type === 'string' && event.type.startsWith('collection.')
+}
+
 // ============================================================================
 // Types
 // ============================================================================

src/env.ts

@@ -99,14 +99,4 @@ export interface Env extends WebhookEnv {
   /** Max request body size in bytes for query endpoint (default: 65536 = 64KB) */
   MAX_QUERY_BODY_SIZE?: string
 
-  // ============================================================================
-  // DeltaLake Integration Configuration
-  // ============================================================================
-
-  /** Enable DeltaTable for event storage (default: false) */
-  USE_DELTALAKE?: string
-  /** Enable DeltaTable for CDC storage (default: false) */
-  USE_DELTALAKE_CDC?: string
-  /** Enable dual-write mode during migration - writes to both legacy and DeltaTable (default: false) */
-  DELTALAKE_DUAL_WRITE?: string
 }

src/event-writer-do.ts

@@ -15,7 +15,7 @@
  */
 
 import { DurableObject } from 'cloudflare:workers'
-import { writeEvents, writeEventsDelta, writeEventsDualWrite, ulid, type EventRecord, type WriteResult, type DeltaWriteResult } from './event-writer'
+import { writeEvents, ulid, type EventRecord, type WriteResult } from './event-writer'
 import { recordWriterDOMetric, recordR2WriteMetric, MetricTimer } from './metrics'
 import type { ShardCoordinatorDO } from './shard-coordinator-do'
 import { logger, sanitize, logError, type Logger } from './logger'
@@ -41,7 +41,7 @@ const DEFAULT_DEDUP_TTL_MS = 24 * 60 * 60 * 1000 // 24 hours default TTL for ded
 
 import type { Env as FullEnv } from './env'
 
-export type Env = Pick<FullEnv, 'EVENTS_BUCKET' | 'EVENT_WRITER' | 'ANALYTICS' | 'SHARD_COORDINATOR' | 'USE_DELTALAKE' | 'DELTALAKE_DUAL_WRITE'>
+export type Env = Pick<FullEnv, 'EVENTS_BUCKET' | 'EVENT_WRITER' | 'ANALYTICS' | 'SHARD_COORDINATOR'>
 
 // ============================================================================
 // Configuration
@@ -547,38 +547,15 @@ export class EventWriterDO extends DurableObject<Env> {
 
     const results: WriteResult[] = []
     let totalBytes = 0
-    const useDeltalake = this.env.USE_DELTALAKE === 'true'
-    const useDualWrite = this.env.DELTALAKE_DUAL_WRITE === 'true'
     try {
       // Write each source group to its own prefix
       for (const [source, sourceEvents] of eventsBySource) {
         const writeTimer = new MetricTimer()
-        let result: WriteResult
-        let bytes: number
-
-        if (useDualWrite) {
-          // Dual-write mode: write to both legacy and DeltaTable during migration
-          const dualResult = await writeEventsDualWrite(this.env.EVENTS_BUCKET, source, sourceEvents, this.shardId)
-          result = dualResult.legacy
-          bytes = dualResult.legacy.bytes + dualResult.delta.bytes
-        } else if (useDeltalake) {
-          // DeltaTable-only mode
-          const deltaResult = await writeEventsDelta(this.env.EVENTS_BUCKET, sourceEvents, this.shardId)
-          // Convert delta result to WriteResult format for compatibility
-          result = {
-            key: deltaResult.path,
-            bytes: deltaResult.bytes,
-            events: deltaResult.events,
-            cpuMs: deltaResult.cpuMs,
-          }
-          bytes = deltaResult.bytes
-        } else {
-          // Legacy mode (default)
-          result = await writeEvents(this.env.EVENTS_BUCKET, source, sourceEvents)
-          bytes = result.bytes
-        }
 
-        logger.child({ component: 'EventWriterDO', shard: this.shardId }).info('Flushed', { key: sanitize.id(result.key, 64), mode: useDualWrite ? 'dual' : useDeltalake ? 'deltalake' : 'legacy' })
+        const result = await writeEvents(this.env.EVENTS_BUCKET, source, sourceEvents)
+        const bytes = result.bytes
+
+        logger.child({ component: 'EventWriterDO', shard: this.shardId }).info('Flushed', { key: sanitize.id(result.key, 64) })
         results.push(result)
         totalBytes += bytes
 

src/event-writer.ts

@@ -11,11 +11,8 @@
  */
 
 import { parquetWriteBuffer } from '@dotdo/hyparquet-writer'
-import { DeltaTable, type StorageBackend } from '@dotdo/deltalake'
 import { ulid } from '../core/src/ulid'
-import { sanitizeR2Path, buildSafeR2Path, sanitizePathSegment, InvalidR2PathError } from './utils'
-import { createR2Storage } from '../core/src/deltalake-storage'
-import { createEventsTable, getEventsTablePath, type EventDeltaRecord } from '../core/src/deltalake-factory'
+import { sanitizeR2Path } from './utils'
 
 export { ulid }
 
@@ -159,155 +156,6 @@ export async function writeEvents(
   return { key, events: events.length, bytes: buffer.byteLength, cpuMs }
 }
 
-// ============================================================================
-// DeltaLake Event Writer
-// ============================================================================
-
-/**
- * Result from a DeltaLake write operation.
- */
-export interface DeltaWriteResult {
-  version: number
-  events: number
-  bytes: number
-  path: string
-  cpuMs: number
-}
-
-/**
- * Convert EventRecord to DeltaLake format (EventDeltaRecord).
- */
-function toEventDeltaRecord(event: EventRecord): EventDeltaRecord {
-  return {
-    ts: event.ts,
-    type: event.type,
-    source: event.source ?? null,
-    provider: event.provider ?? null,
-    eventType: event.eventType ?? null,
-    verified: event.verified ?? null,
-    scriptName: event.scriptName ?? null,
-    outcome: event.outcome ?? null,
-    method: event.method ?? null,
-    url: event.url ?? null,
-    statusCode: event.statusCode ?? null,
-    durationMs: event.durationMs ?? null,
-    payload: event.payload ? JSON.stringify(event.payload) : null,
-  }
-}
-
-/**
- * DeltaLake table cache - one table per shard.
- * Maps shardId to DeltaTable instance.
- */
-const deltaTableCache = new Map<string, DeltaTable<EventDeltaRecord>>()
-
-/**
- * Get or create a DeltaTable for a shard.
- */
-function getDeltaTable(
-  storage: StorageBackend,
-  shardId: number
-): DeltaTable<EventDeltaRecord> {
-  const cacheKey = `events-shard-${shardId}`
-  let table = deltaTableCache.get(cacheKey)
-
-  if (!table) {
-    table = createEventsTable(storage, { shardId })
-    deltaTableCache.set(cacheKey, table)
-  }
-
-  return table
-}
-
-/**
- * Write events to R2 using DeltaLake for ACID guarantees.
- *
- * Events are written to a partitioned DeltaTable with transaction log support.
- * Path structure: events/shard={shardId}/_delta_log/...
- *
- * @param bucket - R2 bucket for storage
- * @param events - Events to write
- * @param shardId - Shard ID for partitioning (default: 0)
- * @returns Result with commit version and event count
- *
- * @example
- * ```typescript
- * const result = await writeEventsDelta(env.EVENTS_BUCKET, events, 2)
- * console.log(`Committed version ${result.version} with ${result.events} events`)
- * ```
- */
-export async function writeEventsDelta(
-  bucket: R2Bucket,
-  events: EventRecord[],
-  shardId: number = 0
-): Promise<DeltaWriteResult> {
-  if (events.length === 0) {
-    throw new Error('No events to write')
-  }
-
-  const startCpu = performance.now()
-
-  // Create storage backend from R2 bucket
-  const storage = createR2Storage(bucket)
-
-  // Get or create the DeltaTable for this shard
-  const table = getDeltaTable(storage, shardId)
-
-  // Convert events to DeltaLake format
-  const deltaRecords = events.map(toEventDeltaRecord)
-
-  // Estimate bytes from serialized records (for metrics)
-  const estimatedBytes = deltaRecords.reduce((sum, record) => {
-    return sum + JSON.stringify(record).length
-  }, 0)
-
-  // Write with ACID guarantees
-  const commit = await table.write(deltaRecords)
-
-  const cpuMs = performance.now() - startCpu
-  const tablePath = getEventsTablePath(shardId)
-
-  console.log(`[WRITE-DELTA] shard=${shardId} version=${commit.version} events=${events.length} cpuMs=${cpuMs.toFixed(2)}`)
-
-  return {
-    version: commit.version,
-    events: events.length,
-    bytes: estimatedBytes,
-    path: `${tablePath}/_delta_log/${String(commit.version).padStart(20, '0')}.json`,
-    cpuMs,
-  }
-}
-
-/**
- * Write events using both legacy Parquet and DeltaLake (dual-write mode).
- *
- * This function writes to both storage formats during migration, ensuring
- * data consistency and enabling gradual rollout of DeltaLake.
- *
- * @param bucket - R2 bucket for storage
- * @param prefix - Legacy path prefix
- * @param events - Events to write
- * @param shardId - Shard ID for DeltaLake partitioning
- * @returns Both legacy and DeltaLake results
- */
-export async function writeEventsDualWrite(
-  bucket: R2Bucket,
-  prefix: string,
-  events: EventRecord[],
-  shardId: number = 0
-): Promise<{ legacy: WriteResult; delta: DeltaWriteResult }> {
-  // Write to both in parallel for performance
-  const [legacyResult, deltaResult] = await Promise.all([
-    writeEvents(bucket, prefix, events),
-    writeEventsDelta(bucket, events, shardId),
-  ])
-
-  return {
-    legacy: legacyResult,
-    delta: deltaResult,
-  }
-}
-
 // ============================================================================
 // Event Buffer - For batching events before writing
 // ============================================================================