/**
 * Framework-agnostic sync engine.
 *
 * Reads the local SQLite database (via Drizzle), gathers projects / roots /
 * events / evidence / secret alerts, scans for secrets, then batches,
 * compresses, and uploads everything to the hosted hub via POST /api/upload.
 *
 * Works with any Drizzle SQLite instance (bun:sqlite or libsql).
 */

import {
  brotliCompressSync,
  gzipSync,
  constants as zlibConstants,
} from "node:zlib";
import { and, desc, eq, gte, inArray, lte, or } from "drizzle-orm";
import {
  evidenceMessages,
  evidenceThreads,
  projectRoots,
  projects,
  secretAlerts,
  syncHistory,
  timelineEvents,
} from "../db/schema";
import { getGitRemote } from "../ingestion/discovery";
import { scanAndRedactEvents } from "../secrets/scanner";
import { getDeviceName, getMachineId } from "../utils/machine-id";
import type {
  SyncEventRow,
  SyncOptions,
  SyncResult,
  UploadEventPayload,
  WorkerProgress,
} from "./types";

// ---------------------------------------------------------------------------
// Internal helpers
// ---------------------------------------------------------------------------

function isMissingExtendedColumnError(error: unknown): boolean {
  const message = error instanceof Error ? error.message : String(error);
  return (
    /(no such column|has no column named|unknown column)/i.test(message) &&
    /(provider_id|event_type|external_thread_id|external_message_id|content_hash|raw_ref|auth_mode|attested_bap_id|attested_sig|ingest_nonce)/i.test(
      message,
    )
  );
}

/**
 * Compute the `since` timestamp to send to `/api/rollups/recompute` so the
 * server rebuilds every rollup bucket this sync could have affected.
 *
 * - explicit date range → recompute from `startDate`
 * - incremental sync → recompute from the last checkpoint
 * - full sync (and first-ever / end-date-only syncs) → recompute from the
 *   epoch, so the ENTIRE historical range is rebuilt.
 *
 * Returning `undefined` for full syncs (the old behaviour) made the server
 * fall back to a 24h lookback, so older rollup buckets were never rebuilt and
 * silently dropped out of the rollup-backed hub timeline ("All time" view)
 * even though the raw events were still uploaded. Rollups are recomputed
 * org-wide server-side, so a full sync must cover the full range, not 24h.
 */
export function computeRollupRecomputeSince(args: {
  startDate?: string;
  incrementalSince?: Date | null;
}): string {
  const { startDate, incrementalSince } = args;
  if (startDate) {
    return new Date(`${startDate}T00:00:00`).toISOString();
  }
  if (incrementalSince) {
    return incrementalSince.toISOString();
  }
  return new Date(0).toISOString();
}

function toUploadEvent(event: SyncEventRow): UploadEventPayload {
  const uploadEvent: UploadEventPayload = {
    id: event.id,
    timestamp: event.timestamp?.toISOString() || new Date().toISOString(),
    source: event.source,
    sourceId: event.sourceId,
    projectRootId: event.projectRootId,
    metadata: event.metadata,
    estimatedMinutes: event.estimatedMinutes,
  };

  if ("providerId" in event) uploadEvent.providerId = event.providerId ?? null;
  if ("eventType" in event) uploadEvent.eventType = event.eventType ?? null;
  if ("externalThreadId" in event) {
    uploadEvent.externalThreadId = event.externalThreadId ?? null;
  }
  if ("externalMessageId" in event) {
    uploadEvent.externalMessageId = event.externalMessageId ?? null;
  }
  if ("contentHash" in event)
    uploadEvent.contentHash = event.contentHash ?? null;
  if ("rawRef" in event) uploadEvent.rawRef = event.rawRef ?? null;
  if ("authMode" in event) uploadEvent.authMode = event.authMode ?? null;
  if ("attestedBapId" in event) {
    uploadEvent.attestedBapId = event.attestedBapId ?? null;
  }
  if ("attestedSig" in event)
    uploadEvent.attestedSig = event.attestedSig ?? null;
  if ("ingestNonce" in event) {
    uploadEvent.ingestNonce = event.ingestNonce ?? null;
  }

  return uploadEvent;
}

function compressSyncPayload(payload: unknown): {
  body: Blob;
  encoding: "br" | "gzip";
  compressedBytes: number;
} {
  const json = JSON.stringify(payload);
  const jsonBuffer = Buffer.from(json);

  try {
    const brotli = brotliCompressSync(jsonBuffer, {
      params: {
        [zlibConstants.BROTLI_PARAM_QUALITY]: 4,
      },
    });

    return {
      body: new Blob([brotli]),
      encoding: "br",
      compressedBytes: brotli.byteLength,
    };
  } catch {
    const gzip = gzipSync(jsonBuffer);
    return {
      body: new Blob([gzip]),
      encoding: "gzip",
      compressedBytes: gzip.byteLength,
    };
  }
}

// ---------------------------------------------------------------------------
// Main engine
// ---------------------------------------------------------------------------

/** Adaptive batching constants. */
const MAX_BATCH_EVENTS = 2000;
const MIN_BATCH_EVENTS = 25;
const TARGET_BATCH_BYTES = 1_500_000;
const CONCURRENCY = 4;

/**
 * Run a full sync from a local Drizzle database to the hosted Scribe hub.
 *
 * Progress is reported through `options.onProgress`. The function resolves
 * with the final {@link SyncResult} or throws on fatal errors.
 */
export async function syncToHub(options: SyncOptions): Promise<SyncResult> {
  const {
    db,
    hostUrl: rawHostUrl,
    authToken: rawAuthToken,
    apiKey: rawApiKey,
    organizationId: rawOrganizationId,
    skipSecretScan,
    fullSync,
    startDate,
    endDate,
    onProgress,
  } = options;

  // --- Validation -----------------------------------------------------------

  const hostUrl = rawHostUrl.replace(/\/$/, "");
  const authToken = rawAuthToken?.trim() || null;
  const apiKey = rawApiKey?.trim() || null;
  const explicitOrganizationId = rawOrganizationId?.trim() || null;

  if (!hostUrl) {
    throw new SyncValidationError("Host URL is required");
  }
  if (!authToken && !apiKey) {
    throw new SyncValidationError("Auth token or API key is required");
  }
  if (authToken && apiKey) {
    throw new SyncValidationError(
      "Provide either auth token or API key, not both",
    );
  }

  // Scope sync-history checkpoints by org to avoid cross-team incremental skips.
  const syncHistoryHostKey = explicitOrganizationId
    ? `${hostUrl}#org:${explicitOrganizationId}`
    : hostUrl;

  // Device info
  const deviceId = getMachineId();
  const deviceName = getDeviceName();

  const syncStartTime = Date.now();
  const results: SyncResult = {
    projects: 0,
    events: 0,
    evidenceThreads: 0,
    evidenceMessages: 0,
    secretAlerts: 0,
    secretsRedacted: 0,
    totalBytes: 0,
    totalTimeMs: 0,
    failedBatches: 0,
    failedEvents: 0,
    errors: [],
  };

  // --- Incremental checkpoint -----------------------------------------------

  let incrementalSince: Date | null = null;
  if (!startDate && !endDate && !fullSync) {
    const latestCompletedSync = await db
      .select({ completedAt: syncHistory.completedAt })
      .from(syncHistory)
      .where(
        and(
          eq(syncHistory.hostUrl, syncHistoryHostKey),
          eq(syncHistory.deviceId, deviceId),
          eq(syncHistory.status, "completed"),
        ),
      )
      .orderBy(desc(syncHistory.completedAt))
      .limit(1)
      .then((r: Array<{ completedAt: Date | null }>) => r[0]);

    if (latestCompletedSync?.completedAt) {
      incrementalSince = latestCompletedSync.completedAt;
    }
  }

  // Create sync history record
  const [syncRecord] = await db
    .insert(syncHistory)
    .values({
      hostUrl: syncHistoryHostKey,
      deviceId,
      startedAt: new Date(),
      status: "running",
    })
    .returning();

  try {
    // === GATHER DATA PHASE ===
    onProgress({ type: "start", phase: "gathering", total: 1 });

    const allProjects = await db.select().from(projects);
    const allRoots = await db.select().from(projectRoots);

    // Build date filter conditions
    const dateConditions = [];
    if (startDate) {
      dateConditions.push(
        gte(timelineEvents.timestamp, new Date(`${startDate}T00:00:00`)),
      );
    }
    if (endDate) {
      dateConditions.push(
        lte(timelineEvents.timestamp, new Date(`${endDate}T23:59:59.999`)),
      );
    }
    if (incrementalSince) {
      dateConditions.push(
        or(
          gte(timelineEvents.createdAt, incrementalSince),
          gte(timelineEvents.editedAt, incrementalSince),
          gte(timelineEvents.annotatedAt, incrementalSince),
        ),
      );
    }

    // Prefer extended provider/event fields when present, but stay compatible
    // with local databases that haven't applied the newer schema yet.
    let extendedColumnsAvailable = true;
    let allEvents: SyncEventRow[] = [];
    try {
      allEvents = (await db
        .select({
          id: timelineEvents.id,
          timestamp: timelineEvents.timestamp,
          source: timelineEvents.source,
          sourceId: timelineEvents.sourceId,
          projectRootId: timelineEvents.projectRootId,
          metadata: timelineEvents.metadata,
          estimatedMinutes: timelineEvents.estimatedMinutes,
          providerId: timelineEvents.providerId,
          eventType: timelineEvents.eventType,
          externalThreadId: timelineEvents.externalThreadId,
          externalMessageId: timelineEvents.externalMessageId,
          contentHash: timelineEvents.contentHash,
          rawRef: timelineEvents.rawRef,
          authMode: timelineEvents.authMode,
          attestedBapId: timelineEvents.attestedBapId,
          attestedSig: timelineEvents.attestedSig,
          ingestNonce: timelineEvents.ingestNonce,
        })
        .from(timelineEvents)
        .where(dateConditions.length > 0 ? and(...dateConditions) : undefined)
        .orderBy(desc(timelineEvents.timestamp))) as SyncEventRow[];
    } catch (error) {
      if (!isMissingExtendedColumnError(error)) {
        throw error;
      }
      extendedColumnsAvailable = false;
      allEvents = (await db
        .select({
          id: timelineEvents.id,
          timestamp: timelineEvents.timestamp,
          source: timelineEvents.source,
          sourceId: timelineEvents.sourceId,
          projectRootId: timelineEvents.projectRootId,
          metadata: timelineEvents.metadata,
          estimatedMinutes: timelineEvents.estimatedMinutes,
        })
        .from(timelineEvents)
        .where(dateConditions.length > 0 ? and(...dateConditions) : undefined)
        .orderBy(desc(timelineEvents.timestamp))) as SyncEventRow[];
    }

    // Gather secret alerts (always send all — they're small and important).
    // Join the linked event's sourceId: local event IDs are device-scoped,
    // so the hub re-resolves the event via (sourceId, deviceId) on upload.
    const allAlerts = await db
      .select({
        id: secretAlerts.id,
        eventId: secretAlerts.eventId,
        eventSourceId: timelineEvents.sourceId,
        type: secretAlerts.type,
        field: secretAlerts.field,
        preview: secretAlerts.preview,
        redacted: secretAlerts.redacted,
        acknowledged: secretAlerts.acknowledged,
        detectedAt: secretAlerts.detectedAt,
      })
      .from(secretAlerts)
      .leftJoin(timelineEvents, eq(secretAlerts.eventId, timelineEvents.id));

    // Gather evidence threads/messages (incremental based on sync history)
    let allEvidenceThreads: Array<typeof evidenceThreads.$inferSelect> = [];
    let allEvidenceMessages: Array<typeof evidenceMessages.$inferSelect> = [];

    if (incrementalSince) {
      allEvidenceThreads = await db
        .select()
        .from(evidenceThreads)
        .where(gte(evidenceThreads.createdAt, incrementalSince));

      if (allEvidenceThreads.length > 0) {
        const threadIds = allEvidenceThreads.map((t: { id: string }) => t.id);
        const BATCH_SIZE = 500;
        for (let i = 0; i < threadIds.length; i += BATCH_SIZE) {
          const batch = threadIds.slice(i, i + BATCH_SIZE);
          const msgs = await db
            .select()
            .from(evidenceMessages)
            .where(inArray(evidenceMessages.threadId, batch));
          allEvidenceMessages.push(...msgs);
        }
      }
    } else {
      allEvidenceThreads = await db.select().from(evidenceThreads);
      allEvidenceMessages = await db.select().from(evidenceMessages);
    }

    const scopeLabel =
      startDate || endDate
        ? "date range"
        : incrementalSince
          ? "incremental"
          : "full";
    onProgress({
      type: "progress",
      phase: "gathering",
      current: `Found ${allProjects.length} projects, ${allRoots.length} roots, ${allEvents.length.toLocaleString()} events, ${allAlerts.length} alerts, ${allEvidenceThreads.length} threads (${scopeLabel}${extendedColumnsAvailable ? ", extended" : ", legacy"})`,
      synced: 1,
      total: 1,
      totalEvents: allEvents.length,
    });

    // === SECRET SCANNING & REDACTION PHASE ===
    const eventsToSync = allEvents;
    if (!skipSecretScan && allEvents.length > 0) {
      const scanStartTime = Date.now();
      onProgress({
        type: "start",
        phase: "scanning",
        total: allEvents.length,
        totalEvents: allEvents.length,
      });

      const scanResult = scanAndRedactEvents(
        allEvents.map((e) => ({ id: e.id, metadata: e.metadata })),
        (scanned, total, findings) => {
          onProgress({
            type: "scanning",
            scanned,
            total,
            findings,
            elapsedMs: Date.now() - scanStartTime,
          });
        },
      );

      if (scanResult.hasSecrets) {
        onProgress({
          type: "secrets_found",
          findings: scanResult.findings,
          totalEvents: allEvents.length,
          redacted: true,
        });
        results.secretsRedacted = scanResult.findings.length;
      }

      onProgress({
        type: "phase_complete",
        phase: "scanning",
        synced: allEvents.length,
        elapsedMs: Date.now() - scanStartTime,
      });
    }

    // === PROJECTS PHASE — send topology once before events ===
    const projectsPhaseStart = Date.now();
    const projectsPayload = allProjects.map(
      (p: { id: string; name: string; color: string | null }) => ({
        id: p.id,
        name: p.name,
        color: p.color,
      }),
    );
    const rootsPayload = await Promise.all(
      allRoots.map(
        async (r: {
          id: string;
          projectId: string | null;
          path: string;
          detectedName: string | null;
          gitRemote: string | null;
        }) => {
          const gitRemote =
            r.gitRemote ?? (await getGitRemote(r.path).catch(() => undefined));

          return {
            id: r.id,
            projectId: r.projectId,
            path: r.path,
            detectedName: r.detectedName,
            gitRemote: gitRemote ?? null,
          };
        },
      ),
    );

    // Build auth headers (shared across all phases)
    const uploadHeadersBase: Record<string, string> = {
      "Content-Type": "application/json",
    };
    if (authToken) {
      uploadHeadersBase.Authorization = `Bearer ${authToken}`;
    }
    if (apiKey) {
      uploadHeadersBase["x-api-key"] = apiKey;
    }
    if (explicitOrganizationId) {
      uploadHeadersBase["x-scribe-organization-id"] = explicitOrganizationId;
    }

    const uploadPayload = async (payload: unknown) => {
      const { body, encoding, compressedBytes } = compressSyncPayload(payload);
      const res = await fetch(`${hostUrl}/api/upload`, {
        method: "POST",
        headers: {
          ...uploadHeadersBase,
          "Content-Encoding": encoding,
        },
        body,
      });
      return { res, compressedBytes };
    };

    let bytesUploaded = 0;

    onProgress({
      type: "start",
      phase: "projects",
      total: allProjects.length + allRoots.length,
    });

    // Send projects/roots as a dedicated batch (no events)
    const projectsOnlyPayload = {
      deviceId,
      deviceName,
      machineId: deviceId,
      projects: projectsPayload,
      projectRoots: rootsPayload,
      events: [],
    };

    try {
      const { res, compressedBytes } = await uploadPayload(projectsOnlyPayload);
      if (res.ok) {
        const result = await res.json();
        results.projects = result.imported?.projects || allProjects.length;
        bytesUploaded += compressedBytes;
      } else {
        const errorBody = await res.text().catch(() => "");
        const uploadError = toSyncUploadError(
          res.status,
          errorBody,
          "Projects upload failed",
        );
        if (
          uploadError.code === "NO_TEAM" ||
          uploadError.code === "SUBSCRIPTION_REQUIRED"
        ) {
          await db
            .update(syncHistory)
            .set({ status: "failed", error: uploadError.code })
            .where(eq(syncHistory.id, syncRecord.id));
        }
        // Topology upload is fatal — events sent without project/root mappings
        // will have projects: [] and projectRoots: [], so events won't be
        // associated with projects. Abort the sync.
        throw uploadError;
      }
    } catch (err) {
      if (err instanceof SyncUploadError) throw err;
      // Topology upload is fatal — events sent without project/root mappings
      // will have projects: [] and projectRoots: [], so events won't be
      // associated with projects. Abort the sync.
      throw new SyncUploadError(
        `Topology upload failed: ${String(err).slice(0, 200)}. Aborting sync — events require project/root mappings.`,
        "TOPOLOGY_UPLOAD_FAILED",
      );
    }

    onProgress({
      type: "phase_complete",
      phase: "projects",
      synced: allProjects.length + allRoots.length,
      elapsedMs: Date.now() - projectsPhaseStart,
    });

    // === EVENTS PHASE ===
    const uploadStartTime = Date.now();

    // Calculate estimated upload size via sampling
    const sampleSize = Math.min(eventsToSync.length, 500);
    const sampleBatch = eventsToSync.slice(0, sampleSize);
    const samplePayload = JSON.stringify({
      events: sampleBatch.map(toUploadEvent),
    });
    const bytesPerEvent =
      sampleBatch.length > 0 ? samplePayload.length / sampleBatch.length : 500;
    const adaptiveBatchSize = Math.floor(
      TARGET_BATCH_BYTES / Math.max(bytesPerEvent, 1),
    );
    const batchSize = Math.min(
      MAX_BATCH_EVENTS,
      Math.max(
        MIN_BATCH_EVENTS,
        adaptiveBatchSize > 0 ? adaptiveBatchSize : MIN_BATCH_EVENTS,
      ),
    );
    const totalBatches = Math.ceil(eventsToSync.length / batchSize);
    const totalEstimatedBytes = Math.round(bytesPerEvent * eventsToSync.length);

    onProgress({
      type: "start",
      phase: "events",
      total: totalBatches,
      totalEvents: eventsToSync.length,
      dataSize: totalEstimatedBytes,
    });

    // Build all batches upfront
    const batches: Array<{
      index: number;
      events: typeof eventsToSync;
      estimatedBytes: number;
    }> = [];

    for (let i = 0; i < eventsToSync.length; i += batchSize) {
      const batchEvents = eventsToSync.slice(i, i + batchSize);
      batches.push({
        index: batches.length,
        events: batchEvents,
        estimatedBytes: Math.round(bytesPerEvent * batchEvents.length),
      });
    }

    let completedBatches = 0;
    let eventsSynced = 0;
    const totalEvents = eventsToSync.length;

    // All event batches send empty projects/roots — topology was sent in the projects phase
    const uploadBatch = async (batch: (typeof batches)[0]) => {
      const payload = {
        deviceId,
        deviceName,
        machineId: deviceId,
        projects: [],
        projectRoots: [],
        events: batch.events.map(toUploadEvent),
      };

      try {
        const { res, compressedBytes } = await uploadPayload(payload);

        if (res.ok) {
          const result = await res.json();
          return {
            success: true as const,
            // Use ?? not ||: a duplicate-only batch legitimately imports 0
            // events, and `0 || batch.events.length` would wrongly report the
            // whole batch as imported — which downstream triggers a full rollup
            // recompute from the epoch. Only a truly-absent count falls back.
            events: result.imported?.events ?? batch.events.length,
            bytes: compressedBytes,
            batchEvents: batch.events.length,
          };
        }
        const errorBody = await res.text().catch(() => "");
        return {
          success: false as const,
          events: 0,
          bytes: 0,
          batchEvents: batch.events.length,
          status: res.status,
          errorBody: errorBody.slice(0, 500),
        };
      } catch (err) {
        return {
          success: false as const,
          events: 0,
          bytes: 0,
          batchEvents: batch.events.length,
          status: 0,
          errorBody: String(err).slice(0, 500),
        };
      }
    };

    // Process all event batches in parallel chunks of CONCURRENCY
    const workerStates: WorkerProgress[] = Array.from(
      { length: CONCURRENCY },
      (_, i) => ({
        workerId: i,
        batchIndex: -1,
        events: 0,
        status: "pending" as const,
      }),
    );

    for (let i = 0; i < batches.length; i += CONCURRENCY) {
      const chunk = batches.slice(i, i + CONCURRENCY);

      for (let w = 0; w < chunk.length; w++) {
        workerStates[w] = {
          workerId: w,
          batchIndex: i + w + 1,
          events: chunk[w].events.length,
          status: "uploading",
        };
      }
      for (let w = chunk.length; w < CONCURRENCY; w++) {
        workerStates[w] = {
          workerId: w,
          batchIndex: -1,
          events: 0,
          status: "pending",
        };
      }

      onProgress({
        type: "progress",
        phase: "events",
        current: `Workers uploading batches ${i + 1}-${Math.min(i + CONCURRENCY, totalBatches)}`,
        synced: completedBatches,
        total: totalBatches,
        eventsSynced,
        totalEvents,
        elapsedMs: Date.now() - uploadStartTime,
        bytesUploaded,
        totalBytes: totalEstimatedBytes,
        workers: [...workerStates],
      });

      const chunkResults = await Promise.all(chunk.map(uploadBatch));

      for (let w = 0; w < chunkResults.length; w++) {
        const result = chunkResults[w];
        completedBatches++;
        workerStates[w].status = result.success ? "done" : "error";
        if (result.success) {
          eventsSynced += result.events;
          bytesUploaded += result.bytes;
          results.events += result.events;
        } else {
          const batchNumber = i + w + 1;
          results.failedBatches++;
          results.failedEvents += result.batchEvents;
          results.errors.push({
            batch: batchNumber,
            status: result.status ?? 0,
            body: result.errorBody ?? "",
          });
          onProgress({
            type: "error",
            phase: "events",
            message: toCleanProgressError(
              result.status ?? 0,
              result.errorBody ?? "",
              `Batch ${batchNumber} failed`,
            ),
          });
        }
      }

      const elapsedMs = Date.now() - uploadStartTime;
      const eventsPerMs = eventsSynced > 0 ? eventsSynced / elapsedMs : 1;
      const remainingEvents = totalEvents - eventsSynced;
      const estimatedRemainingMs = Math.round(remainingEvents / eventsPerMs);

      onProgress({
        type: "progress",
        phase: "events",
        current: `${eventsSynced.toLocaleString()} / ${totalEvents.toLocaleString()} events`,
        synced: completedBatches,
        total: totalBatches,
        eventsSynced,
        totalEvents,
        elapsedMs,
        estimatedRemainingMs,
        bytesUploaded,
        totalBytes: totalEstimatedBytes,
        workers: [...workerStates],
      });
    }

    const uploadElapsedMs = Date.now() - uploadStartTime;
    onProgress({
      type: "phase_complete",
      phase: "events",
      synced: eventsSynced,
      elapsedMs: uploadElapsedMs,
    });

    // === EVIDENCE THREADS PHASE ===
    // Threads must be uploaded before messages (messages reference threadIds).
    // Hoisted so the message phase can access the mapping.
    const clientThreadIdMap = new Map<string, string>();
    const EVIDENCE_THREAD_BATCH_SIZE = 500;
    if (allEvidenceThreads.length > 0) {
      const threadBatches: Array<typeof allEvidenceThreads> = [];
      for (
        let i = 0;
        i < allEvidenceThreads.length;
        i += EVIDENCE_THREAD_BATCH_SIZE
      ) {
        threadBatches.push(
          allEvidenceThreads.slice(i, i + EVIDENCE_THREAD_BATCH_SIZE),
        );
      }

      onProgress({
        type: "start",
        phase: "evidence_threads",
        total: threadBatches.length,
      });

      const threadPhaseStart = Date.now();
      let threadsSynced = 0;

      // Sequential — threads must exist on server before messages reference them
      for (let bi = 0; bi < threadBatches.length; bi++) {
        const batch = threadBatches[bi];
        const payload = {
          deviceId,
          deviceName,
          machineId: deviceId,
          events: [],
          projects: [],
          projectRoots: [],
          evidenceThreads: batch.map(
            (t: {
              id: string;
              providerId: string;
              externalThreadId: string;
              workspacePath: string | null;
              workspaceHash: string | null;
              title: string | null;
              startedAt: Date | null;
              endedAt: Date | null;
              providerMetadata: unknown;
            }) => ({
              id: t.id,
              providerId: t.providerId,
              externalThreadId: t.externalThreadId,
              workspacePath: t.workspacePath,
              workspaceHash: t.workspaceHash,
              title: t.title,
              startedAt: t.startedAt?.toISOString() || null,
              endedAt: t.endedAt?.toISOString() || null,
              providerMetadata: t.providerMetadata,
            }),
          ),
        };

        try {
          const { res, compressedBytes } = await uploadPayload(payload);
          if (res.ok) {
            const result = await res.json();
            results.evidenceThreads +=
              result.imported?.evidenceThreads || batch.length;
            bytesUploaded += compressedBytes;
            threadsSynced += batch.length;
            // Capture local→hub thread ID mapping for message remapping
            if (result.threadIdMap && typeof result.threadIdMap === "object") {
              for (const [localId, hubId] of Object.entries(
                result.threadIdMap,
              )) {
                clientThreadIdMap.set(localId, hubId as string);
              }
            }
          } else {
            const errorBody = await res.text().catch(() => "");
            results.failedBatches++;
            results.errors.push({
              batch: completedBatches + bi + 1,
              status: res.status,
              body: errorBody.slice(0, 500),
            });
            onProgress({
              type: "error",
              phase: "evidence_threads",
              message: toCleanProgressError(
                res.status,
                errorBody,
                `Thread batch ${bi + 1} failed`,
              ),
            });
          }
        } catch (err) {
          results.failedBatches++;
          results.errors.push({
            batch: completedBatches + bi + 1,
            status: 0,
            body: String(err).slice(0, 500),
          });
          onProgress({
            type: "error",
            phase: "evidence_threads",
            message: `Thread batch ${bi + 1} failed: ${String(err).slice(0, 200)}`,
          });
        }

        onProgress({
          type: "progress",
          phase: "evidence_threads",
          current: `${threadsSynced.toLocaleString()} / ${allEvidenceThreads.length.toLocaleString()} threads`,
          synced: bi + 1,
          total: threadBatches.length,
          elapsedMs: Date.now() - threadPhaseStart,
          bytesUploaded,
        });
      }

      onProgress({
        type: "phase_complete",
        phase: "evidence_threads",
        synced: threadsSynced,
        elapsedMs: Date.now() - threadPhaseStart,
      });
    }

    // === EVIDENCE MESSAGES PHASE ===
    const EVIDENCE_MESSAGE_BATCH_SIZE = 2000;
    if (allEvidenceMessages.length > 0) {
      const messageBatches: Array<typeof allEvidenceMessages> = [];
      for (
        let i = 0;
        i < allEvidenceMessages.length;
        i += EVIDENCE_MESSAGE_BATCH_SIZE
      ) {
        messageBatches.push(
          allEvidenceMessages.slice(i, i + EVIDENCE_MESSAGE_BATCH_SIZE),
        );
      }

      onProgress({
        type: "start",
        phase: "evidence_messages",
        total: messageBatches.length,
      });

      const messagePhaseStart = Date.now();
      let messagesSynced = 0;

      const uploadMessageBatch = async (
        batch: typeof allEvidenceMessages,
        _batchIndex: number,
      ) => {
        // Remap message threadIds from local→hub using mapping captured in thread phase.
        // Drop messages whose threads weren't mapped (thread batch failed).
        const mappedMessages = batch.filter((m: { threadId: string }) =>
          clientThreadIdMap.has(m.threadId),
        );
        if (mappedMessages.length < batch.length) {
          const dropped = batch.length - mappedMessages.length;
          onProgress({
            type: "error",
            phase: "evidence_messages",
            message: `Dropped ${dropped} messages with unmapped thread IDs`,
          });
        }
        if (mappedMessages.length === 0) {
          return { success: true, messages: 0, bytes: 0 };
        }
        const payload = {
          deviceId,
          deviceName,
          machineId: deviceId,
          events: [],
          projects: [],
          projectRoots: [],
          evidenceMessages: mappedMessages.map(
            (m: {
              id: string;
              threadId: string;
              providerId: string;
              externalMessageId: string | null;
              role: string;
              timestamp: Date;
              contentText: string | null;
              contentHash: string | null;
              toolName: string | null;
              toolStatus: string | null;
              tokenIn: number | null;
              tokenOut: number | null;
            }) => {
              // mappedMessages was produced by `.filter(m => clientThreadIdMap.has(m.threadId))`,
              // so this lookup always hits.
              const threadId = clientThreadIdMap.get(m.threadId);
              if (!threadId) {
                throw new Error(
                  `invariant: message ${m.id} lost its thread mapping after filter`,
                );
              }
              return {
                id: m.id,
                threadId,
                providerId: m.providerId,
                externalMessageId: m.externalMessageId,
                role: m.role,
                timestamp: m.timestamp.toISOString(),
                contentText: m.contentText,
                contentHash: m.contentHash,
                toolName: m.toolName,
                toolStatus: m.toolStatus,
                tokenIn: m.tokenIn,
                tokenOut: m.tokenOut,
              };
            },
          ),
        };

        try {
          const { res, compressedBytes } = await uploadPayload(payload);
          if (res.ok) {
            const result = await res.json();
            return {
              success: true as const,
              count: result.imported?.evidenceMessages || batch.length,
              bytes: compressedBytes,
            };
          }
          const errorBody = await res.text().catch(() => "");
          return {
            success: false as const,
            count: 0,
            bytes: 0,
            status: res.status,
            errorBody: errorBody.slice(0, 500),
          };
        } catch (err) {
          return {
            success: false as const,
            count: 0,
            bytes: 0,
            status: 0,
            errorBody: String(err).slice(0, 500),
          };
        }
      };

      // Parallel with CONCURRENCY workers
      for (let i = 0; i < messageBatches.length; i += CONCURRENCY) {
        const chunk = messageBatches.slice(i, i + CONCURRENCY);
        const chunkResults = await Promise.all(
          chunk.map((batch, w) => uploadMessageBatch(batch, i + w)),
        );

        for (let w = 0; w < chunkResults.length; w++) {
          const result = chunkResults[w];
          if (result.success) {
            messagesSynced += result.count;
            bytesUploaded += result.bytes;
            results.evidenceMessages += result.count;
          } else {
            results.failedBatches++;
            results.errors.push({
              batch: completedBatches + i + w + 1,
              status: result.status ?? 0,
              body: result.errorBody ?? "",
            });
            onProgress({
              type: "error",
              phase: "evidence_messages",
              message: toCleanProgressError(
                result.status ?? 0,
                result.errorBody ?? "",
                `Message batch ${i + w + 1} failed`,
              ),
            });
          }
        }

        onProgress({
          type: "progress",
          phase: "evidence_messages",
          current: `${messagesSynced.toLocaleString()} / ${allEvidenceMessages.length.toLocaleString()} messages`,
          synced: Math.min(i + CONCURRENCY, messageBatches.length),
          total: messageBatches.length,
          elapsedMs: Date.now() - messagePhaseStart,
          bytesUploaded,
        });
      }

      onProgress({
        type: "phase_complete",
        phase: "evidence_messages",
        synced: messagesSynced,
        elapsedMs: Date.now() - messagePhaseStart,
      });
    }

    // === SECRET ALERTS PHASE ===
    if (allAlerts.length > 0) {
      onProgress({
        type: "start",
        phase: "secret_alerts",
        total: 1,
      });

      const alertPhaseStart = Date.now();
      const alertPayload = {
        deviceId,
        deviceName,
        machineId: deviceId,
        events: [],
        projects: [],
        projectRoots: [],
        secretAlerts: allAlerts.map(
          (a: {
            id: string;
            eventId: string | null;
            eventSourceId: string | null;
            type: string;
            field: string;
            preview: string;
            redacted: boolean;
            acknowledged: boolean;
            detectedAt: Date | null;
          }) => ({
            id: a.id,
            eventId: a.eventId,
            eventSourceId: a.eventSourceId,
            type: a.type,
            field: a.field,
            preview: a.preview,
            redacted: a.redacted,
            acknowledged: a.acknowledged,
            detectedAt: a.detectedAt?.toISOString() || null,
          }),
        ),
      };

      try {
        const { res, compressedBytes } = await uploadPayload(alertPayload);
        if (res.ok) {
          const result = await res.json();
          results.secretAlerts =
            result.imported?.secretAlerts || allAlerts.length;
          bytesUploaded += compressedBytes;
        } else {
          const errorBody = await res.text().catch(() => "");
          results.failedBatches++;
          results.errors.push({
            batch: completedBatches + 1,
            status: res.status,
            body: errorBody.slice(0, 500),
          });
          onProgress({
            type: "error",
            phase: "secret_alerts",
            message: toCleanProgressError(
              res.status,
              errorBody,
              "Secret alerts upload failed",
            ),
          });
        }
      } catch (err) {
        results.failedBatches++;
        results.errors.push({
          batch: completedBatches + 1,
          status: 0,
          body: String(err).slice(0, 500),
        });
        onProgress({
          type: "error",
          phase: "secret_alerts",
          message: `Secret alerts upload failed: ${String(err).slice(0, 200)}`,
        });
      }

      onProgress({
        type: "phase_complete",
        phase: "secret_alerts",
        synced: allAlerts.length,
        elapsedMs: Date.now() - alertPhaseStart,
      });
    }

    // === ROLLUPS PHASE — trigger server-side rollup recomputation ===
    if (results.events > 0) {
      const rollupPhaseStart = Date.now();
      onProgress({
        type: "start",
        phase: "rollups",
        total: 1,
      });

      try {
        const rollupRes = await fetch(`${hostUrl}/api/rollups/recompute`, {
          method: "POST",
          headers: {
            ...uploadHeadersBase,
            "Content-Type": "application/json",
          },
          body: JSON.stringify({
            // Recompute every rollup bucket this sync could have affected.
            // Full syncs recompute from the epoch so the entire historical
            // range is rebuilt — omitting `since` made the server fall back to
            // a 24h lookback, silently dropping older buckets from the
            // rollup-backed hub timeline. See computeRollupRecomputeSince.
            since: computeRollupRecomputeSince({ startDate, incrementalSince }),
          }),
        });

        if (!rollupRes.ok) {
          const errorBody = await rollupRes.text().catch(() => "");
          onProgress({
            type: "error",
            phase: "rollups",
            message: toCleanProgressError(
              rollupRes.status,
              errorBody,
              "Rollup recomputation failed",
            ),
          });
        }
      } catch (err) {
        onProgress({
          type: "error",
          phase: "rollups",
          message: `Rollup recomputation failed: ${String(err).slice(0, 200)}`,
        });
      }

      onProgress({
        type: "phase_complete",
        phase: "rollups",
        synced: results.events,
        elapsedMs: Date.now() - rollupPhaseStart,
      });
    }

    results.totalBytes = bytesUploaded;
    results.totalTimeMs = Date.now() - syncStartTime;

    // Update sync history record — mark partial if any batches failed
    const finalStatus = results.failedBatches > 0 ? "partial" : "completed";
    await db
      .update(syncHistory)
      .set({
        completedAt: new Date(),
        eventsUploaded: results.events,
        projectsUploaded: results.projects,
        totalBytes: results.totalBytes,
        status: finalStatus,
        ...(results.failedBatches > 0
          ? {
              error: `${results.failedBatches} batch(es) failed (${results.failedEvents} events)`,
            }
          : {}),
      })
      .where(eq(syncHistory.id, syncRecord.id));

    onProgress({ type: "complete", results });

    return results;
  } catch (error) {
    // Mark sync as failed in history
    await db
      .update(syncHistory)
      .set({
        status: "failed",
        error:
          error instanceof SyncUploadError &&
          (error.code === "NO_TEAM" || error.code === "SUBSCRIPTION_REQUIRED")
            ? error.code
            : String(error),
      })
      .where(eq(syncHistory.id, syncRecord.id))
      .catch(() => {}); // Best-effort — don't mask original error

    throw error;
  }
}

// ---------------------------------------------------------------------------
// Error classes
// ---------------------------------------------------------------------------

/** Thrown for invalid options before any work begins. */
export class SyncValidationError extends Error {
  constructor(message: string) {
    super(message);
    this.name = "SyncValidationError";
  }
}

/** Thrown when the hub returns a fatal upload error (e.g. NO_TEAM). */
export class SyncUploadError extends Error {
  code: string;
  upgradeUrl?: string;

  constructor(message: string, code: string, upgradeUrl?: string) {
    super(message);
    this.name = "SyncUploadError";
    this.code = code;
    this.upgradeUrl = upgradeUrl;
  }
}

interface ParsedSyncErrorBody {
  code?: string;
  message?: string;
  upgradeUrl?: string;
}

function parseSyncErrorBody(rawBody: string): ParsedSyncErrorBody {
  try {
    const parsed: unknown = JSON.parse(rawBody);
    if (!parsed || typeof parsed !== "object") return {};

    const body = parsed as Record<string, unknown>;
    const details =
      body.details && typeof body.details === "object"
        ? (body.details as Record<string, unknown>)
        : null;
    const code =
      typeof body.error === "string"
        ? body.error
        : typeof body.code === "string"
          ? body.code
          : undefined;

    return {
      code,
      message: typeof body.message === "string" ? body.message : undefined,
      upgradeUrl:
        typeof details?.upgradeUrl === "string"
          ? details.upgradeUrl
          : undefined,
    };
  } catch {
    return {};
  }
}

/** Convert a failed hub upload response into a safe, structured error. */
export function toSyncUploadError(
  status: number,
  rawBody: string,
  context: string,
): SyncUploadError {
  const parsed = parseSyncErrorBody(rawBody);

  if (parsed.code === "NO_TEAM") {
    return new SyncUploadError(
      "You must create or join a team on the hosted hub before syncing.",
      "NO_TEAM",
    );
  }

  if (parsed.code === "SUBSCRIPTION_REQUIRED" || status === 402) {
    return new SyncUploadError(
      parsed.message || "Multi-device sync requires a paid plan.",
      "SUBSCRIPTION_REQUIRED",
      parsed.upgradeUrl,
    );
  }

  return new SyncUploadError(
    `${context}: ${parsed.message || `sync failed (HTTP ${status})`}`,
    "UPLOAD_FAILED",
  );
}

function toCleanProgressError(
  status: number,
  rawBody: string,
  context: string,
): string {
  const parsed = parseSyncErrorBody(rawBody);
  return parsed.message
    ? `${context}: ${parsed.message}`
    : `${context} (HTTP ${status})`;
}