Asynchronous Compaction & Drift Prevention

// Generic merge engine. Works with any entity type that has a
// checksum, version vector, and typed fields.

export type MergeStrategy = "last_writer_wins" | "field_level" | "time_ranked" | "manual_review";

export interface MergePlan {
  canonical: object;
  mergeLog: MergeEntry[];
  strategy: MergeStrategy;
}

export interface MergeEntry {
  field: string;
  resolution: "kept" | "replaced" | "merged" | "filled";
  fromValues: unknown[];
  toValue: unknown;
  reason: string;
}

export class MergeEngine {
  merge<T extends { updated_at: number; checksum?: string }>(
    duplicates: T[],
    strategy: MergeStrategy,
  ): MergePlan {
    switch (strategy) {
      case "last_writer_wins":
        return this.lastWriterWins(duplicates);
      case "field_level":
        return this.fieldLevelReconciliation(duplicates);
      case "time_ranked":
        return this.timeRankedComposite(duplicates);
      case "manual_review":
        return { canonical: duplicates[0], mergeLog: [], strategy };
      default:
        return this.lastWriterWins(duplicates);
    }
  }

  private lastWriterWins<T extends { updated_at: number }>(
    duplicates: T[],
  ): MergePlan {
    const sorted = [...duplicates].sort(
      (a, b) => b.updated_at - a.updated_at,
    );
    const canonical = { ...sorted[0] };
    const mergeLog: MergeEntry[] = [];

    // For each field, take the most recently written value;
    // for arrays, union all values.
    for (const key of Object.keys(canonical)) {
      if (key === "updated_at" || key === "checksum") continue;

      const values = sorted.map((d) => (d as Record<string, unknown>)[key]);
      const isArray = Array.isArray(values[0]);

      if (isArray) {
        const merged = [...new Set(values.flat() as unknown[])];
        (canonical as Record<string, unknown>)[key] = merged;
        mergeLog.push({
          field: key,
          resolution: "merged",
          fromValues: values,
          toValue: merged,
          reason: `Set union of ${values.length} duplicates`,
        });
      } else {
        const winner = values[0];
        mergeLog.push({
          field: key,
          resolution: values.slice(1).every((v) => v === winner) ? "kept" : "replaced",
          fromValues: values,
          toValue: winner,
          reason: `Last writer wins: selected from most recent node`,
        });
      }
    }

    return { canonical, mergeLog, strategy: "last_writer_wins" };
  }

  private fieldLevelReconciliation<T extends { updated_at: number }>(
    duplicates: T[],
  ): MergePlan {
    const base = {} as Record<string, unknown>;
    const mergeLog: MergeEntry[] = [];
    const entries = Object.keys(duplicates[0] as Record<string, unknown>)
      .filter((k) => k !== "checksum");

    for (const key of entries) {
      const values = duplicates.map((d) =>
        (d as Record<string, unknown>)[key],
      );
      const unique = [...new Set(values.map((v) => JSON.stringify(v)))].map(
        (s) => JSON.parse(s),
      );

      if (unique.length === 1) {
        base[key] = unique[0];
        continue;
      }

      // Resolve by field type
      const sample = unique[0];
      if (typeof sample === "number" && key.endsWith("_count")) {
        // Counter field: sum
        base[key] = (unique as number[]).reduce((a, b) => a + b, 0);
        mergeLog.push({
          field: key,
          resolution: "merged",
          fromValues: values,
          toValue: base[key],
          reason: `Counter: summed ${unique.length} values`,
        });
      } else if (Array.isArray(sample)) {
        // Array field: union
        base[key] = [...new Set(unique.flat() as unknown[])];
        mergeLog.push({
          field: key,
          resolution: "merged",
          fromValues: values,
          toValue: base[key],
          reason: `Array: set union of ${unique.length} sources`,
        });
      } else if (key === "updated_at") {
        // Timestamp: pick max
        base[key] = Math.max(...(unique as number[]));
      } else {
        // Scalar: pick most recent node's value
        const mostRecent = [...duplicates].sort(
          (a, b) => b.updated_at - a.updated_at,
        )[0];
        base[key] = (mostRecent as Record<string, unknown>)[key];
        mergeLog.push({
          field: key,
          resolution: "replaced",
          fromValues: values,
          toValue: base[key],
          reason: `Scalar: last writer wins`,
        });
      }
    }

    return {
      canonical: base as T,
      mergeLog,
      strategy: "field_level",
    };
  }

  private timeRankedComposite<T extends { updated_at: number }>(
    duplicates: T[],
  ): MergePlan {
    // Rank by: populated fields (desc), edge count (desc), recency (desc)
    const ranked = [...duplicates].sort((a, b) => {
      const aPopulated = Object.values(a as Record<string, unknown>)
        .filter((v) => v !== null && v !== undefined && v !== "").length;
      const bPopulated = Object.values(b as Record<string, unknown>)
        .filter((v) => v !== null && v !== undefined && v !== "").length;
      if (aPopulated !== bPopulated) return bPopulated - aPopulated;
      return b.updated_at - a.updated_at;
    });

    const base = { ...ranked[0] } as Record<string, unknown>;
    const mergeLog: MergeEntry[] = [];

    // Fill gaps from lower-ranked nodes
    for (let i = 1; i < ranked.length; i++) {
      for (const [key, value] of Object.entries(ranked[i] as Record<string, unknown>)) {
        if (key === "checksum" || key === "updated_at") continue;
        const current = base[key];
        if (current === null || current === undefined || current === "" || 
            (Array.isArray(current) && current.length === 0)) {
          base[key] = value;
          mergeLog.push({
            field: key,
            resolution: "filled",
            fromValues: [current, value],
            toValue: value,
            reason: `Gap-filled from rank-${i} node`,
          });
        }
      }
    }

    return { canonical: base as T, mergeLog, strategy: "time_ranked" };
  }
}

// Full reconciliation worker. Runs as a background cron that
// iterates over entities, fetches source metadata, re-plays the
// pipeline, and verifies consistency.

export class ReconciliationWorker {
  private stores: EntityStores;
  private graphUpdater: GraphUpdater;
  private snapshotManager: SnapshotManager;

  constructor(
    stores: EntityStores,
    graphUpdater: GraphUpdater,
    snapshotManager: SnapshotManager,
  ) {
    this.stores = stores;
    this.graphUpdater = graphUpdater;
    this.snapshotManager = snapshotManager;
  }

  async reconcileEntity(
    entityType: string,
    entityId: string,
  ): Promise<ReconciliationResult> {
    // 1. Fetch the current entity state
    const current = await this.stores.getEntity(entityType, entityId);
    if (!current) {
      return { status: "not_found", entityType, entityId };
    }

    // 2. Compute current checksum
    const currentChecksum = computeEntityChecksum(current);
    const storedChecksum = (current as Record<string, unknown>).checksum as string;

    // 3. Compare checksums
    if (currentChecksum === storedChecksum) {
      return { status: "consistent", entityType, entityId };
    }

    // 4. Checksum mismatch - fetch source metadata and replay
    const sourceEvents = await this.fetchSourceEvents(entityType, entityId);
    if (sourceEvents.length === 0) {
      // No source data: update the stored checksum to match reality
      await this.stores.updateChecksum(entityType, entityId, currentChecksum);
      return {
        status: "checksum_corrected",
        entityType,
        entityId,
        previousChecksum: storedChecksum,
        newChecksum: currentChecksum,
      };
    }

    // 5. Take a snapshot before re-merging
    const compactionId = `reconcile:${entityType}:${entityId}:${Date.now()}`;
    await this.snapshotManager.takeSnapshot(entityType, current, compactionId);

    // 6. Re-merge from source events
    let merged = current;
    let changed = false;

    for (const event of sourceEvents) {
      const result = await this.graphUpdater.processEmail(event);
      if (result.status === "applied") {
        changed = true;
      }
      // Re-fetch the entity after each merge
      merged = await this.stores.getEntity(entityType, entityId);
    }

    // 7. Recompute checksum after merge
    const finalChecksum = merged
      ? computeEntityChecksum(merged)
      : currentChecksum;

    if (changed) {
      await this.stores.updateChecksum(entityType, entityId, finalChecksum);
      return {
        status: "reconciled",
        entityType,
        entityId,
        previousChecksum: storedChecksum,
        newChecksum: finalChecksum,
        eventsReplayed: sourceEvents.length,
      };
    }

    return {
      status: "checksum_updated",
      entityType,
      entityId,
      previousChecksum: storedChecksum,
      newChecksum: finalChecksum,
    };
  }

  async reconcileAll(
    entityTypes: string[],
    concurrency = 4,
  ): Promise<ReconciliationSummary> {
    const summary: ReconciliationSummary = {
      total: 0,
      consistent: 0,
      reconciled: 0,
      errors: [],
    };

    for (const entityType of entityTypes) {
      const ids = await this.stores.listEntityIds(entityType);
      summary.total += ids.length;

      // Process with limited concurrency
      const batches = this.chunk(ids, concurrency);
      for (const batch of batches) {
        const results = await Promise.allSettled(
          batch.map((id) => this.reconcileEntity(entityType, id)),
        );

        for (const result of results) {
          if (result.status === "fulfilled") {
            if (result.value.status === "consistent") {
              summary.consistent++;
            } else if (
              result.value.status === "reconciled" ||
              result.value.status === "checksum_corrected"
            ) {
              summary.reconciled++;
            }
          } else {
            summary.errors.push(result.reason?.message ?? "unknown");
          }
        }
      }
    }

    return summary;
  }

  private async fetchSourceEvents(
    entityType: string,
    entityId: string,
  ): Promise<EmailMetadataEvent[]> {
    // Query the source mail store for all events involving this entity
    const source = this.stores.getSourceStore(entityType);
    if (!source) return [];
    return source.queryByEntity(entityType, entityId);
  }

  private chunk<T>(arr: T[], size: number): T[][] {
    const chunks: T[][] = [];
    for (let i = 0; i < arr.length; i += size) {
      chunks.push(arr.slice(i, i + size));
    }
    return chunks;
  }
}

export type ReconciliationResult =
  | { status: "consistent"; entityType: string; entityId: string }
  | { status: "not_found"; entityType: string; entityId: string }
  | { status: "checksum_corrected"; entityType: string; entityId: string; previousChecksum: string; newChecksum: string }
  | { status: "reconciled"; entityType: string; entityId: string; previousChecksum: string; newChecksum: string; eventsReplayed: number }
  | { status: "checksum_updated"; entityType: string; entityId: string; previousChecksum: string; newChecksum: string }
  | { status: "error"; entityType: string; entityId: string; error: string };

interface ReconciliationSummary {
  total: number;
  consistent: number;
  reconciled: number;
  errors: string[];
}