Add HashResolver to handle collection defaults via SHA1 hash resolution and enhance migration progress tracking

- Introduced `HashResolver` to map drill_events{sha1} collections to (appId, eventName) pairs.
- Updated schema and config to include `countlyDb` for hash resolution.
- Integrated `HashResolver` into Orchestrator for resolving collection-level defaults.
- Enhanced progress tracking with estimated document counts and collection-level stats in HTTP stats route.
- Refined document transformation to leverage collection hash defaults for `a` and `e` fields.

Author: kanwarujjaval

src/config/loader.ts

@@ -22,6 +22,7 @@ function envToRawConfig(env: NodeJS.ProcessEnv) {
         source: {
             uri: env.MONGO_URI,
             db: env.MONGO_DB,
+            countlyDb: env.MONGO_COUNTLY_DB,
             collectionPrefix: env.MONGO_COLLECTION_PREFIX,
             readPreference: env.MONGO_READ_PREFERENCE,
             readConcern: env.MONGO_READ_CONCERN,

src/config/schema.ts

@@ -49,6 +49,7 @@ export const configSchema = z.object({
     source: z.object({
         uri: z.string().min(1),
         db: z.string().default("countly_drill"),
+        countlyDb: z.string().default("countly"),
         collectionPrefix: z.string().default("drill_events"),
         readPreference: z.string().default("primary"),
         readConcern: z.string().default("majority"),

src/http/stats-route.ts

@@ -8,6 +8,35 @@ import type { CommandFlags } from '../state/redis-hot-state.ts';
 import type { Config } from '../config/schema.ts';
 import type { CollectionOrchestrator, OrchestratorProgress } from '../runtime/collection-orchestrator.ts';
 
+// ─────────────────────────────────────────────────────────────────────────────
+// Formatting helpers
+// ─────────────────────────────────────────────────────────────────────────────
+
+function progressBar(pct: number, width = 30): string {
+  const filled = Math.round((pct / 100) * width);
+  const arrow = filled < width ? '>' : '';
+  const empty = Math.max(0, width - filled - (arrow ? 1 : 0));
+  return '[' + '='.repeat(filled) + arrow + ' '.repeat(empty) + '] ' + pct + '%';
+}
+
+function fmtDuration(ms: number): string {
+  const s = Math.floor(ms / 1000);
+  const h = Math.floor(s / 3600);
+  const m = Math.floor((s % 3600) / 60);
+  const sec = s % 60;
+  if (h > 0) return `${h}h ${m}m ${sec}s`;
+  if (m > 0) return `${m}m ${sec}s`;
+  return `${sec}s`;
+}
+
+function fmtNum(n: number): string {
+  return n.toLocaleString('en-US');
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Route
+// ─────────────────────────────────────────────────────────────────────────────
+
 export interface StatsDeps {
   orchestrator: CollectionOrchestrator;
   mongoReader: { isConnected(): boolean };
@@ -46,6 +75,7 @@ export function registerStatsRoute(app: FastifyInstance, deps: StatsDeps): void
     const batchStats = orchestrator.getStats();
     const runnerStatus = orchestrator.getStatus();
     const currentBatchSeq = orchestrator.getCurrentBatchSeq();
+    const estimatedCounts = orchestrator.getEstimatedCounts();
 
     // Find the current run ID from the orchestrator's progress
     const currentResult = progress.results.find(
@@ -78,6 +108,27 @@ export function registerStatsRoute(app: FastifyInstance, deps: StatsDeps): void
     const elapsedMs = batchStats?.elapsedMs ?? 0;
     const elapsedSec = elapsedMs / 1000;
 
+    // ── Progress calculations ───────────────────────────────────────────
+    const currentCollEstimate = progress.currentCollection
+      ? estimatedCounts.get(progress.currentCollection) ?? 0 : 0;
+    const currentCollDocsRead = totalDocsRead;
+    const currentCollPct = currentCollEstimate > 0
+      ? Math.min(100, Math.round((currentCollDocsRead / currentCollEstimate) * 100)) : 0;
+
+    // Overall: completed collections use actual docsRead, fallback to estimate
+    const totalEstimated = Array.from(estimatedCounts.values()).reduce((a, b) => a + b, 0);
+    const completedDocsRead = progress.results
+      .filter(r => r.status === 'completed' || r.status === 'skipped')
+      .reduce((sum, r) => sum + (r.docsRead ?? estimatedCounts.get(r.collection) ?? 0), 0);
+    const overallDocsRead = completedDocsRead + currentCollDocsRead;
+    const overallPct = totalEstimated > 0
+      ? Math.min(100, Math.round((overallDocsRead / totalEstimated) * 100)) : 0;
+
+    // ETA based on elapsed time and progress percentage
+    const etaMs = overallPct > 0 && overallPct < 100 && elapsedMs > 0
+      ? Math.round(((Date.now() - startedAt.getTime()) / overallPct) * (100 - overallPct))
+      : null;
+
     // Commands from Redis (best-effort)
     let commands: CommandFlags = {};
     if (runId) {
@@ -91,6 +142,37 @@ export function registerStatsRoute(app: FastifyInstance, deps: StatsDeps): void
     const redisMetrics = redisState.getMetrics();
 
     const payload = {
+      // ── Quick-glance summary ──────────────────────────────────────────
+      summary: {
+        overall: progressBar(overallPct),
+        overallPct,
+        currentCollection: progress.currentCollection
+          ? `${progress.currentCollection}  ${progressBar(currentCollPct)}`
+          : 'idle',
+        currentCollectionPct: currentCollPct,
+        collections: `${progress.completedCollections}/${progress.totalCollections} done`
+          + (progress.failedCollections > 0 ? `, ${progress.failedCollections} failed` : '')
+          + (progress.skippedCollections > 0 ? `, ${progress.skippedCollections} skipped` : ''),
+        docsProgress: `${fmtNum(overallDocsRead)} / ~${fmtNum(totalEstimated)} docs`,
+        throughput: `${fmtNum(Math.round(batchStats?.docsPerSecond ?? 0))} docs/s`,
+        elapsed: fmtDuration(Date.now() - startedAt.getTime()),
+        eta: etaMs !== null ? `~${fmtDuration(etaMs)}` : 'calculating...',
+        status: runnerStatus,
+      },
+
+      // ── Current collection detail ─────────────────────────────────────
+      currentCollectionProgress: progress.currentCollection ? {
+        collection: progress.currentCollection,
+        estimated: currentCollEstimate,
+        docsRead: currentCollDocsRead,
+        rowsInserted: totalRowsInserted,
+        pct: currentCollPct,
+        bar: progressBar(currentCollPct),
+        batchSeq: currentBatchSeq,
+        skipRate: currentCollDocsRead > 0
+          ? `${((totalDocsSkipped / currentCollDocsRead) * 100).toFixed(1)}%` : '0%',
+      } : null,
+
       service: {
         name: config.service.name,
         version,
@@ -107,6 +189,13 @@ export function registerStatsRoute(app: FastifyInstance, deps: StatsDeps): void
         skippedCollections: progress.skippedCollections,
         currentCollection: progress.currentCollection,
         collections: progress.collections,
+        collectionProgress: progress.results.map(r => ({
+          collection: r.collection,
+          status: r.status,
+          estimated: estimatedCounts.get(r.collection) ?? null,
+          docsRead: r.docsRead ?? null,
+          rowsInserted: r.rowsInserted ?? null,
+        })),
       },
       run: {
         sourceNs: runRecord?.source_ns ?? null,

src/main.ts

@@ -8,6 +8,7 @@ import { MongoReader } from './source/mongo-reader.ts';
 import { ClickHouseWriter } from './target/clickhouse-writer.ts';
 import { ClickHousePressure } from './target/clickhouse-pressure.ts';
 import { CollectionOrchestrator } from './runtime/collection-orchestrator.ts';
+import { HashResolver } from './transform/hash-resolver.ts';
 import { RetryPolicy } from './runtime/retry-policy.ts';
 import { GcController, type GcConfig } from './runtime/gc-controller.ts';
 import { ProcessMetricsCollector } from './runtime/process-metrics.ts';
@@ -119,6 +120,14 @@ async function main(): Promise<void> {
   }
   logger.info('All external services connected');
 
+  // ── 4b. Build hash resolver for drill_events* collection defaults ────
+  const hashResolver = new HashResolver(
+    { uri: config.source.uri, countlyDb: config.source.countlyDb },
+    logger,
+  );
+  await hashResolver.build();
+  logger.info({ hashEntries: hashResolver.size }, 'Hash resolver built');
+
   // ── 5. Create CollectionOrchestrator ────────────────────────────────
   const orchestrator = new CollectionOrchestrator({
     manifestStore,
@@ -128,6 +137,7 @@ async function main(): Promise<void> {
     chPressure,
     gcController,
     retryPolicy,
+    hashResolver,
     logger,
     config,
   });
@@ -227,6 +237,7 @@ async function main(): Promise<void> {
     await closeResource('ClickHouse', async () => { await chWriter.close(); await pressureClient.close(); });
     await closeResource('Redis', () => redisState.close());
     await closeResource('ManifestStore', () => manifestStore.close());
+    await closeResource('HashResolver', () => hashResolver.close());
     processMetrics.stop();
     gcController.dispose();
 

src/runtime/batch-runner.ts

@@ -12,6 +12,7 @@ import type { GcController } from "./gc-controller.ts";
 import type { RetryPolicy } from "./retry-policy.ts";
 import { SkipCounter, type SkipReason } from "../transform/skip-reasons.ts";
 import { transformBatch, type OutputRow } from "../transform/normalize.ts";
+import type { CollectionDefaults } from "../transform/hash-resolver.ts";
 import { type Cursor, deserializeCursor, serializeCursor } from "../types/cursor.ts";
 
 // ---------------------------------------------------------------------------
@@ -30,6 +31,7 @@ export interface BatchRunnerConfig {
   database: string;
   table: string;
   snapshotInterval: number;
+  collectionDefaults?: CollectionDefaults;
 }
 
 export interface BatchRunnerDeps {
@@ -283,6 +285,7 @@ export class BatchRunner {
         const { rows, skippedSamples } = transformBatch(
           page.docs,
           this.skipCounter,
+          this.deps.config.collectionDefaults,
         );
 
         // Record skip samples in manifest
@@ -732,7 +735,7 @@ export class BatchRunner {
         );
 
         // Step 2: Re-transform
-        const { rows } = transformBatch(page.docs, this.skipCounter);
+        const { rows } = transformBatch(page.docs, this.skipCounter, this.deps.config.collectionDefaults);
 
         // Step 3: Recompute digest
         const newDigest = computePayloadDigest(rows);

src/runtime/collection-orchestrator.ts

@@ -9,6 +9,7 @@ import type { RetryPolicy } from "./retry-policy.ts";
 import { BatchRunner, type RunnerStatus, type BatchRunnerStats } from "./batch-runner.ts";
 import { resolveRun } from "./resolve-run.ts";
 import { discoverCollections } from "../source/discover-collections.ts";
+import type { HashResolver } from "../transform/hash-resolver.ts";
 import type { Config } from "../config/schema.ts";
 
 // ---------------------------------------------------------------------------
@@ -21,6 +22,8 @@ export interface CollectionResult {
     runId: string;
     status: "completed" | "failed" | "skipped";
     error?: string;
+    docsRead?: number;
+    rowsInserted?: number;
 }
 
 export interface OrchestratorResult {
@@ -48,6 +51,7 @@ export interface OrchestratorDeps {
     chPressure: ClickHousePressure;
     gcController: GcController;
     retryPolicy: RetryPolicy;
+    hashResolver: HashResolver;
     logger: Logger;
     config: Config;
 }
@@ -62,6 +66,7 @@ export class CollectionOrchestrator {
 
     private discoveredCollections: string[] = [];
     private results: CollectionResult[] = [];
+    private estimatedCounts: Map<string, number> = new Map();
     private currentCollection: string | null = null;
     private currentRunId: string | null = null;
     private currentBatchRunner: BatchRunner | null = null;
@@ -221,6 +226,10 @@ export class CollectionOrchestrator {
         return this.currentBatchRunner?.getCurrentBatchSeq() ?? 0;
     }
 
+    getEstimatedCounts(): Map<string, number> {
+        return new Map(this.estimatedCounts);
+    }
+
     // -----------------------------------------------------------------------
     // Private
     // -----------------------------------------------------------------------
@@ -237,6 +246,29 @@ export class CollectionOrchestrator {
         // Switch MongoReader to this collection (also ensures index)
         await mongoReader.switchCollection(collectionName);
 
+        // Get estimated document count for progress tracking
+        const estimatedCount = await mongoReader.getEstimatedCount();
+        this.estimatedCounts.set(collectionName, estimatedCount);
+        this.logger.info({ collection: collectionName, estimatedCount }, "Estimated document count");
+
+        // Resolve collection defaults from hash (once per collection)
+        const collectionDefaults = this.deps.hashResolver.resolveCollectionName(
+            collectionName,
+            config.source.collectionPrefix,
+        );
+
+        if (collectionDefaults) {
+            this.logger.info(
+                { collection: collectionName, appId: collectionDefaults.a, event: collectionDefaults.e },
+                "Resolved collection hash defaults",
+            );
+        } else if (collectionName !== config.source.collectionPrefix) {
+            this.logger.warn(
+                { collection: collectionName },
+                "No hash match found for collection — documents missing a/e will be skipped",
+            );
+        }
+
         // Create per-collection RedisHotState with isolated key prefix
         const redisClient = this.deps.redisState.getRedisClient();
         const collectionPrefix = `${config.state.redisKeyPrefix}:${collectionName}`;
@@ -291,6 +323,7 @@ export class CollectionOrchestrator {
                 database: config.target.db,
                 table: config.target.table,
                 snapshotInterval: config.state.timelineSnapshotInterval,
+                collectionDefaults: collectionDefaults ?? undefined,
             },
         });
 
@@ -301,19 +334,30 @@ export class CollectionOrchestrator {
             await batchRunner.run();
 
             const finalStatus = batchRunner.getStatus();
+            const finalStats = batchRunner.getStats();
             if (finalStatus === "completed") {
-                return { collection: collectionName, sourceNs, runId, status: "completed" };
+                return {
+                    collection: collectionName, sourceNs, runId, status: "completed",
+                    docsRead: finalStats.totalDocsRead,
+                    rowsInserted: finalStats.totalRowsInserted,
+                };
             }
             if (finalStatus === "stopped" || finalStatus === "stopping") {
                 // Stopped by operator — treat as incomplete, not failed
-                return { collection: collectionName, sourceNs, runId, status: "failed", error: "Stopped by operator" };
+                return {
+                    collection: collectionName, sourceNs, runId, status: "failed", error: "Stopped by operator",
+                    docsRead: finalStats.totalDocsRead,
+                    rowsInserted: finalStats.totalRowsInserted,
+                };
             }
             return {
                 collection: collectionName,
                 sourceNs,
                 runId,
                 status: "failed",
                 error: `BatchRunner ended with status: ${finalStatus}`,
+                docsRead: finalStats.totalDocsRead,
+                rowsInserted: finalStats.totalRowsInserted,
             };
         } catch (err) {
             const error = err instanceof Error ? err.message : String(err);