/**
 * Instance Sync Manager — P2P data replication between Crow instances.
 *
 * Separate from the contact-to-contact SyncManager. Instance sync replicates
 * core data (memories, context, contacts, etc.) between instances owned by
 * the same user, using Hypercore feeds and Lamport timestamps.
 *
 * Entry format: { table, op, row, lamport_ts, instance_id, signature }
 *
 * Conflict resolution: if incoming.lamport_ts > local.lamport_ts → apply;
 * otherwise run the equivalence check — equal data → skip silently; differing
 * data → log to sync_conflicts and notify the operator.
 */

import Hypercore from "hypercore";
import { mkdirSync, existsSync } from "node:fs";
import { resolve } from "node:path";
import { sign, verify } from "./identity.js";
import { resolveDataDir } from "../db.js";
import { isSyncable } from "../gateway/dashboard/settings/sync-allowlist.js";
import bus from "../shared/event-bus.js";

/**
 * Build the canonical wire row for a crow_context DB row.
 *
 * Allowlist: {section_key, section_title, content, sort_order, enabled,
 *   device_id, project_id}.  Never includes id (instance-local AUTOINCREMENT),
 * lamport_ts, or updated_at.  enabled comes from the DB row as an INTEGER (0/1)
 * — better-sqlite3 throws on boolean binds, so the caller must not coerce it.
 *
 * Used by every crow_context emit site so the allowlist is single-sourced.
 *
 * @param {object} dbRow — a row returned by a SELECT * FROM crow_context query
 * @returns {object} wire object safe to pass to emitChange
 */
export function buildCrowContextWireRow(dbRow) {
  return {
    section_key:   dbRow.section_key,
    section_title: dbRow.section_title,
    content:       dbRow.content,
    sort_order:    dbRow.sort_order,
    enabled:       dbRow.enabled,  // INTEGER 0/1 from DB
    device_id:     dbRow.device_id   ?? null,
    project_id:    dbRow.project_id  ?? null,
  };
}

// Tables that participate in core sync
export const SYNCED_TABLES = [
  "memories",
  "crow_context",
  "contacts",
  "shared_items",
  "messages",
  "relay_config",
  "crow_instances",
  // Phase 5-full: LLM provider registry. Operator-editable on any instance;
  // changes propagate to peers so model swaps/additions are one-and-done.
  // Toggle off by removing this line if you want per-instance divergence.
  "providers",
  // LLM consolidation: per-agent provider/model overrides for presets.
  // Synthetic `id` PK (= "${preset_name}:${agent_name}") so the standard
  // _applyInsert / _applyUpdate / _applyDelete paths dispatch correctly.
  "orchestrator_role_overrides",
  // Scoped-settings replication (2026-04-14): only rows with instance_id IS NULL
  // AND whose key is in the SYNC_ALLOWLIST are actually emitted/applied.
  // See shouldSyncRow() below.
  "dashboard_settings",
  // Phase 6 (meta-glasses note-taking). research_notes + glasses_note_sessions
  // replicate so notes captured via glasses surface on paired instances.
  // research_projects stays local-only — see OUTBOUND_TRANSFORMS: research_notes
  // below, which NULLs project_id on the wire.
  "research_notes",
  "glasses_note_sessions",
];

// Columns to exclude from sync payloads (security-sensitive or instance-local)
const EXCLUDED_COLUMNS = {
  crow_instances: ["auth_token_hash"],
  // apiKey can be sensitive for some deployments; keep in sync by default for
  // local-lab scenarios but flag here as the place to exclude if paranoid.
  providers: [],
};

// Per-table outbound mutations applied right after the EXCLUDED_COLUMNS strip.
// Use for cases where the local representation differs from the synced
// representation — e.g. research_notes.project_id is a local FK (project rows
// aren't synced), so we NULL it on the wire. Synced notes show up on peers
// as project-less (null project_id).
const OUTBOUND_TRANSFORMS = {
  research_notes: (row) => ({ ...row, project_id: null }),
};

/**
 * Equivalence check: incoming row's values match the local row on every key
 * present in the incoming row.
 *
 * Per-key rules: both null/undefined → equal; exactly one nullish → NOT equal
 * (never alias null with ""); otherwise String(a) === String(b).
 * lamport_ts and instance_id are ignored — sync metadata, not content.
 *
 * "a" is the LOCAL row (already passed through OUTBOUND_TRANSFORMS at the call
 * site). "b" is the INCOMING (wire) row. Only keys in "b" are checked: wire
 * entries are partial rows (only changed or relevant fields), so keys absent
 * from the incoming row are not part of the equivalence judgement.
 *
 * @param {object} a - Local row (pre-transformed for wire comparison)
 * @param {object} b - Incoming (wire) row
 * @returns {boolean}
 */
export function rowsEquivalent(a, b) {
  if (!a || !b) return false;
  const IGNORE = new Set(["lamport_ts", "instance_id"]);
  for (const k of Object.keys(b)) {
    if (IGNORE.has(k)) continue;
    const av = a[k] ?? null;
    const bv = b[k] ?? null;
    const aNullish = av === null || av === undefined;
    const bNullish = bv === null || bv === undefined;
    if (aNullish && bNullish) continue;            // both absent/null → equal
    if (aNullish !== bNullish) return false;        // exactly one nullish → not equal
    if (String(av) !== String(bv)) return false;
  }
  return true;
}

/**
 * Per-table filter: decide whether a mutated row should be broadcast (or an
 * inbound row applied). Returning false means "this row is local-only; skip it".
 *
 * For dashboard_settings we enforce:
 *   - instance_id IS NULL (i.e. the global scope row)
 *   - key matches the SYNC_ALLOWLIST
 *
 * All other synced tables default to true (no filter).
 */
// Debounced storage-client invalidation. The six storage.shared.* keys arrive
// as six separate sync entries within milliseconds; a 500ms trailing debounce
// collapses that burst into one resetStorageClient() call after the write
// settles, avoiding five intermediate rebuilds against partial config.
let _storageResetTimer = null;
function _scheduleStorageReset() {
  if (_storageResetTimer) clearTimeout(_storageResetTimer);
  _storageResetTimer = setTimeout(async () => {
    _storageResetTimer = null;
    try {
      const { resetStorageClient } = await import("../storage/s3-client.js");
      resetStorageClient();
    } catch (err) {
      console.warn("[instance-sync] resetStorageClient import failed:", err.message);
    }
  }, 500);
}

function shouldSyncRow(table, row) {
  if (table !== "dashboard_settings") return true;
  if (!row || !row.key) return false;
  // dashboard_settings holds only the global scope; per-instance overrides live
  // in dashboard_settings_overrides (never synced). Allowlist gates the key.
  return isSyncable(row.key);
}

export class InstanceSyncManager {
  /**
   * @param {object} identity - Crow identity (from loadOrCreateIdentity)
   * @param {import("@libsql/client").Client} db
   * @param {string} localInstanceId - This instance's UUID
   */
  constructor(identity, db, localInstanceId) {
    this.identity = identity;
    this.db = db;
    this.localInstanceId = localInstanceId;
    this.dataDir = resolve(resolveDataDir(), "instance-sync");

    // Per-remote-instance paired feeds
    this.outFeeds = new Map(); // remoteInstanceId → Hypercore (we write)
    this.inFeeds = new Map();  // remoteInstanceId → Hypercore (they write)

    // Per-peer serialization for initInstance() — see initInstance() for why.
    this._initLocks = new Map(); // remoteInstanceId → tail Promise

    // Per-peer serialization for _processNewEntries — see D4 fix below.
    this._processLocks = new Map(); // remoteInstanceId → tail Promise

    // Flag that _ensureCounter has seeded the sync_state row at least once.
    // The DB row is the authority; this is only a cheap short-circuit.
    this._counterSeeded = false;

    // Whether the manager has been started
    this.started = false;
  }

  /**
   * Idempotent seeder: ensure a sync_state row exists for this instance.
   * INSERT OR IGNORE is atomic against concurrent first-callers; safe to race.
   * All three sync_state writers call this before their UPDATE so that a fresh
   * receive-only instance (which never calls _nextLamport) still has a row to
   * UPDATE — without it the counter advances and checkpoints silently no-op.
   */
  async _ensureCounter() {
    if (this._counterSeeded) return;
    try {
      await this.db.execute({
        sql: "INSERT OR IGNORE INTO sync_state (instance_id, local_counter) VALUES (?, 0)",
        args: [this.localInstanceId],
      });
      // Only latch on success — a transient DB error must not permanently
      // disable seeding (the writers would then no-op forever after recovery).
      this._counterSeeded = true;
    } catch (err) {
      console.warn("[instance-sync] Failed to seed sync_state row:", err.message);
    }
  }

  /**
   * Atomically increment and persist the local Lamport counter.
   * Uses a single UPDATE ... RETURNING so no two concurrent callers can race
   * on the same value — better-sqlite3 is synchronous, each statement is atomic
   * with respect to all JS interleavings.
   * @returns {Promise<number>} The new counter value
   */
  async _nextLamport() {
    await this._ensureCounter();
    const { rows } = await this.db.execute({
      sql: `UPDATE sync_state SET local_counter = local_counter + 1, updated_at = datetime('now')
            WHERE instance_id = ? RETURNING local_counter`,
      args: [this.localInstanceId],
    });
    if (rows.length === 0) {
      // Seed race: the INSERT OR IGNORE above ran but the UPDATE found no row
      // (another caller's INSERT lost the race). Seed explicitly and retry once.
      this._counterSeeded = false;
      await this._ensureCounter();
      const retry = await this.db.execute({
        sql: `UPDATE sync_state SET local_counter = local_counter + 1, updated_at = datetime('now')
              WHERE instance_id = ? RETURNING local_counter`,
        args: [this.localInstanceId],
      });
      if (!retry.rows[0]) {
        // Seed + retry both failed — the DB is genuinely unavailable. Throw
        // with a real message instead of a bare TypeError out of emitChange.
        throw new Error("[instance-sync] sync_state row missing after seed+retry — DB unavailable?");
      }
      return Number(retry.rows[0].local_counter);
    }
    return Number(rows[0].local_counter);
  }

  /**
   * Advance the local counter so it is greater than an incoming Lamport value.
   * Single atomic MAX(...) UPDATE — no read-check-write race across an await.
   * @param {number} incomingTs - Lamport timestamp from a remote entry
   */
  async _advanceCounter(incomingTs) {
    await this._ensureCounter();
    await this.db.execute({
      sql: `UPDATE sync_state SET local_counter = MAX(local_counter, CAST(? AS INTEGER) + 1), updated_at = datetime('now')
            WHERE instance_id = ?`,
      args: [Number(incomingTs) || 0, this.localInstanceId],
    });
  }

  /**
   * Initialize feeds for a remote instance.
   * @param {string} remoteInstanceId
   * @param {Buffer|null} theirFeedKey - The remote instance's outgoing feed key (null if unknown yet)
   */
  async initInstance(remoteInstanceId, theirFeedKey) {
    // Serialize per-peer to prevent concurrent fd-lock contention on
    // <feed-dir>/db/LOCK. Multiple startup paths converge here for the
    // same peer (boot loop at server.js, eagerInitPairedPeers, the
    // onInstanceConnected callback when Hyperswarm connects, tailnet-sync,
    // onInstanceKeyReceived after the handshake). The outFeeds.has()/
    // inFeeds.has() guards alone are not atomic across the await in
    // Hypercore.ready(), so two concurrent callers would each construct
    // a second Hypercore on the same on-disk feed, and the loser would
    // throw "File descriptor could not be locked".
    //
    // We can't cache the first call's promise and short-circuit subsequent
    // calls, because a later caller may arrive with a real theirFeedKey
    // that a prior null-key call skipped — we need to open the inFeed on
    // the later turn. So we strictly chain instead: each call awaits the
    // prior call, then re-evaluates state.
    const prior = this._initLocks.get(remoteInstanceId) || Promise.resolve();
    const next = prior
      .catch(() => {}) // a failed prior turn shouldn't block our attempt
      .then(() => this._initInstanceInner(remoteInstanceId, theirFeedKey));
    this._initLocks.set(remoteInstanceId, next);
    try {
      return await next;
    } finally {
      // If we're still the tail, drop the entry so the Map doesn't retain
      // a settled promise per peer over the process lifetime.
      if (this._initLocks.get(remoteInstanceId) === next) {
        this._initLocks.delete(remoteInstanceId);
      }
    }
  }

  async _initInstanceInner(remoteInstanceId, theirFeedKey) {
    const dir = resolve(this.dataDir, remoteInstanceId);
    mkdirSync(resolve(dir, "out"), { recursive: true });
    mkdirSync(resolve(dir, "in"), { recursive: true });

    // Our outgoing feed (writable by us)
    if (!this.outFeeds.has(remoteInstanceId)) {
      const outFeed = new Hypercore(resolve(dir, "out"), {
        valueEncoding: "json",
      });
      await outFeed.ready();
      this.outFeeds.set(remoteInstanceId, outFeed);
    }

    // Their incoming feed (writable by them)
    if (!this.inFeeds.has(remoteInstanceId) && theirFeedKey) {
      const inFeed = new Hypercore(resolve(dir, "in"), theirFeedKey, {
        valueEncoding: "json",
      });
      await inFeed.ready();

      // Listen for new entries
      inFeed.on("append", async () => {
        await this._processNewEntries(remoteInstanceId, inFeed);
      });

      this.inFeeds.set(remoteInstanceId, inFeed);
    }

    return this.outFeeds.get(remoteInstanceId);
  }

  /**
   * Eagerly open outbound feeds for every paired peer in `crow_instances`.
   * MUST be called at gateway boot BEFORE any `emitChange` can fire — otherwise
   * early emissions (memory stores, settings writes at startup, bundle
   * registrations) vanish silently: `emitChange` iterates `outFeeds`, which is
   * empty until a peer WebSocket lands, so its `feed.append` loop has zero
   * iterations while the Lamport counter advances anyway.
   *
   * Safe to re-call; `initInstance` no-ops if the out-feed already exists.
   * In-feeds are also pre-opened if the peer's `sync_url` (their outbound
   * feed key) is already cached — this lets startup `_processNewEntries`
   * catch up entries that mirrored during a previous session.
   */
  async eagerInitPairedPeers() {
    let rows = [];
    try {
      const r = await this.db.execute({
        sql: "SELECT id, sync_url FROM crow_instances WHERE status IN ('active','offline') AND id != ?",
        args: [this.localInstanceId],
      });
      rows = r.rows || [];
    } catch (err) {
      console.warn(`[instance-sync] eagerInitPairedPeers: db query failed: ${err.message}`);
      return 0;
    }
    let opened = 0;
    for (const peer of rows) {
      try {
        const theirKey = peer.sync_url ? Buffer.from(peer.sync_url, "hex") : null;
        await this.initInstance(peer.id, theirKey);
        // Catch up any in-feed entries that arrived during a past session
        // but were never processed because `append` only fires on NEW
        // replicated blocks, not on re-opening an existing feed.
        const inFeed = this.inFeeds.get(peer.id);
        if (inFeed && inFeed.length > 0) {
          this._processNewEntries(peer.id, inFeed).catch((err) =>
            console.warn(`[instance-sync] eager catch-up failed for ${peer.id.slice(0,12)}…: ${err.message}`),
          );
        }
        opened++;
      } catch (err) {
        console.warn(`[instance-sync] eager init failed for ${peer.id.slice(0,12)}…: ${err.message}`);
      }
    }
    if (opened > 0) {
      console.log(`[instance-sync] eagerly opened ${opened} peer feed(s) — emit pipeline armed`);
    }
    return opened;
  }

  /**
   * One-shot reconciliation: re-emit every sync-allowlisted dashboard_settings
   * row so that peers whose pre-fix outFeed dropped entries can catch up.
   * Guarded by a flag row so it runs once per instance lifetime (post-fix).
   *
   * Called at boot AFTER eagerInitPairedPeers. Idempotent: re-running is a
   * no-op. On the peer side, `_applyDashboardSetting` skips rows whose
   * lamport_ts is stale or whose value is unchanged, so the re-emit is
   * safe even when the peer already has current data.
   */
  async reemitSyncableSettingsOnce() {
    const FLAG_KEY = "__sync_reemit_allowlist_v1";
    let alreadyRan = false;
    try {
      const { rows } = await this.db.execute({
        sql: "SELECT value FROM dashboard_settings WHERE key = ?",
        args: [FLAG_KEY],
      });
      alreadyRan = rows?.length > 0;
    } catch {}
    if (alreadyRan) return 0;

    if (this.outFeeds.size === 0) {
      // No paired peers — nothing to reconcile with. Still mark done so we
      // don't keep checking on every boot.
      try {
        await this.db.execute({
          sql: "INSERT INTO dashboard_settings (key, value, updated_at) VALUES (?, 'no-peers', datetime('now')) ON CONFLICT(key) DO NOTHING",
          args: [FLAG_KEY],
        });
      } catch {}
      return 0;
    }

    // dashboard_settings is the global scope only — per-instance scopes live
    // in dashboard_settings_overrides (never synced). No instance_id column.
    let rows;
    try {
      const r = await this.db.execute({
        sql: "SELECT key, value, lamport_ts FROM dashboard_settings",
      });
      rows = r.rows || [];
    } catch (err) {
      console.warn(`[instance-sync] reemit read failed: ${err.message}`);
      return 0;
    }

    let emitted = 0;
    for (const row of rows) {
      if (!isSyncable(row.key)) continue;
      try {
        await this.emitChange("dashboard_settings", "update", {
          key: row.key,
          value: row.value,
          instance_id: null,
        });
        emitted++;
      } catch (err) {
        console.warn(`[instance-sync] reemit ${row.key} failed: ${err.message}`);
      }
    }

    try {
      await this.db.execute({
        sql: "INSERT INTO dashboard_settings (key, value, updated_at) VALUES (?, ?, datetime('now')) ON CONFLICT(key) DO NOTHING",
        args: [FLAG_KEY, `done:${emitted}`],
      });
    } catch {}

    if (emitted > 0) {
      console.log(`[instance-sync] one-shot re-emit: ${emitted} sync-allowlisted setting(s) → peers will reconcile`);
    }
    return emitted;
  }

  /**
   * Get the local outgoing feed key for a remote instance.
   * Used during instance handshake so the remote knows our feed key.
   */
  getOutFeedKey(remoteInstanceId) {
    const feed = this.outFeeds.get(remoteInstanceId);
    return feed ? feed.key : null;
  }

  /**
   * Replicate feeds over a NoiseSecretStream-wrapped transport. Hyperswarm
   * connections are already NoiseSecretStream instances; for plain WebSocket
   * or other Duplex transports, wrap with `new NoiseSecretStream(isInitiator,
   * underlyingDuplex)` first (see tailnet-sync.js).
   *
   * @param {string} remoteInstanceId
   * @param {object} stream - NoiseSecretStream (Hypercore reads .noiseStream from it)
   */
  async replicate(remoteInstanceId, stream) {
    const outFeed = this.outFeeds.get(remoteInstanceId);
    const inFeed = this.inFeeds.get(remoteInstanceId);

    if (outFeed) outFeed.replicate(stream, { live: true });
    if (inFeed) inFeed.replicate(stream, { live: true });
  }

  /**
   * Emit a sync entry for a local data change.
   * Called by memory/context/sharing servers after mutations.
   *
   * @param {string} table - Table name (e.g., "memories")
   * @param {"insert"|"update"|"delete"} op - Operation type
   * @param {object} row - The row data (for insert/update) or { id } (for delete)
   */
  async emitChange(table, op, row) {
    if (!SYNCED_TABLES.includes(table)) return;
    if (!shouldSyncRow(table, row)) return; // local-only row; don't broadcast

    const lamportTs = await this._nextLamport();

    // Strip excluded columns
    const excluded = EXCLUDED_COLUMNS[table] || [];
    let cleanRow = { ...row };
    for (const col of excluded) {
      delete cleanRow[col];
    }
    // Apply per-table outbound transform (e.g. NULL project_id on notes).
    const transform = OUTBOUND_TRANSFORMS[table];
    if (transform) cleanRow = transform(cleanRow);

    const entry = {
      table,
      op,
      row: cleanRow,
      lamport_ts: lamportTs,
      instance_id: this.localInstanceId,
    };

    // Sign the entry (signature over JSON without the signature field)
    const payload = JSON.stringify(entry);
    entry.signature = sign(payload, this.identity.ed25519Priv);

    // Update the row's lamport_ts in the local database
    if (op !== "delete") {
      try {
        if (table === "dashboard_settings" && row.key !== undefined) {
          await this.db.execute({
            sql: `UPDATE dashboard_settings SET lamport_ts = ? WHERE key = ?`,
            args: [lamportTs, row.key],
          });
        } else if (table === "crow_context" && row.section_key !== undefined) {
          // Composite-key stamp; MAX(COALESCE(...)) guards against out-of-order
          // concurrent emitChange calls — plain MAX(NULL,x) is NULL in SQLite so
          // the COALESCE is load-bearing.
          await this.db.execute({
            sql: `UPDATE crow_context SET lamport_ts = MAX(COALESCE(lamport_ts, 0), ?)
                  WHERE section_key = ? AND device_id IS ? AND project_id IS ?`,
            args: [lamportTs, row.section_key, row.device_id ?? null, row.project_id ?? null],
          });
        } else if (row.id !== undefined) {
          await this.db.execute({
            sql: `UPDATE ${table} SET lamport_ts = ? WHERE id = ?`,
            args: [lamportTs, row.id],
          });
        }
      } catch {
        // Non-fatal — row may not have lamport_ts column yet
      }
    }

    // Append to all outgoing feeds (broadcast to all connected instances)
    for (const [instanceId, feed] of this.outFeeds) {
      try {
        await feed.append(entry);
      } catch (err) {
        console.warn(`[instance-sync] Failed to append to feed for ${instanceId}:`, err.message);
      }
    }

    return lamportTs;
  }

  /**
   * Public wrapper: serialize per-peer to prevent overlapping runs (D4).
   * Same promise-chain pattern as _initLocks — a failed run's .catch(() => {})
   * ensures the next queued run is not blocked by a prior error.
   * lastSeq is re-read INSIDE the inner run after acquiring the chain, so a
   * queued run naturally no-ops the range the prior run already covered.
   */
  async _processNewEntries(remoteInstanceId, feed) {
    const prior = this._processLocks.get(remoteInstanceId) || Promise.resolve();
    const next = prior
      .catch(() => {})
      .then(() => this._processNewEntriesInner(remoteInstanceId, feed));
    this._processLocks.set(remoteInstanceId, next);
    try {
      return await next;
    } finally {
      if (this._processLocks.get(remoteInstanceId) === next) {
        this._processLocks.delete(remoteInstanceId);
      }
    }
  }

  /**
   * Inner body: process new entries from a remote instance's incoming feed.
   * lastSeq is read here (after acquiring the per-peer chain) so each
   * queued run starts from where the prior run actually left off.
   *
   * Checkpoints AFTER every ATTEMPTED entry (seq + 1, outside the try-catch).
   * This intentionally advances the checkpoint past failed entries — halting
   * on a poison entry would freeze the entire peer feed, a worse outcome than
   * skipping one bad entry. The trailing whole-batch checkpoint is removed.
   */
  async _processNewEntriesInner(remoteInstanceId, feed) {
    // Re-read lastSeq inside the lock — the prior chained run may have advanced it.
    const lastSeq = await this._getLastAppliedSeq(remoteInstanceId);

    for (let seq = lastSeq; seq < feed.length; seq++) {
      try {
        const entry = await feed.get(seq);
        await this._applyEntry(remoteInstanceId, entry);
      } catch (err) {
        // Skip-and-continue: a poison entry must not freeze this peer's feed.
        console.warn(`[instance-sync] Failed to process entry ${seq} from ${remoteInstanceId}:`, err.message);
      }
      // Checkpoint after every attempted entry — outside the try so it advances
      // even when _applyEntry threw (intentional skip-log-advance semantics).
      await this._setLastAppliedSeq(remoteInstanceId, seq + 1);
    }
  }

  /**
   * Apply a single sync entry from a remote instance.
   */
  async _applyEntry(remoteInstanceId, entry) {
    const { table, op, row, lamport_ts, instance_id, signature } = entry;

    // Validate table
    if (!SYNCED_TABLES.includes(table)) return;

    // Defense in depth: drop inbound rows that fail the local syncability check.
    // A peer claiming a non-allowlisted dashboard_settings key (or a row with
    // instance_id != NULL) is either misconfigured or malicious — either way we don't apply.
    if (!shouldSyncRow(table, row)) return;

    // Verify signature
    const entryWithoutSig = { table, op, row, lamport_ts, instance_id };
    const payload = JSON.stringify(entryWithoutSig);

    // Look up the instance's public key
    const verified = verify(payload, signature, this.identity.ed25519Pubkey);
    // Note: For now, all instances share the same identity (same user).
    // In future, verify against the instance's registered public key.

    if (!verified) {
      console.warn(`[instance-sync] Signature verification failed for entry from ${remoteInstanceId}`);
      return;
    }

    // Advance local counter
    await this._advanceCounter(lamport_ts);

    // Tables keyed by 'key' instead of 'id' need special-case apply logic.
    if (table === "dashboard_settings") {
      try {
        await this._applyDashboardSetting(op, row, lamport_ts);
        if (row?.key && String(row.key).startsWith("storage.shared.")) {
          _scheduleStorageReset();
        }
      } catch (err) {
        console.warn(`[instance-sync] Failed to apply ${op} on dashboard_settings:`, err.message);
      }
      return;
    }

    // crow_context is keyed by composite (section_key, device_id, project_id) — route
    // ALL ops (insert/update/delete) through the composite-key handler so that:
    //   • insert collisions get LWW upsert instead of the null-id warn dead-end
    //   • updates and deletes are applied/conflicted by composite key rather than id
    // The invalidateContextCache() call that previously lived below the switch now
    // fires inside _applyCrowContext on every apply, which is the ONLY path reached
    // (the return below prevents fall-through).  The old position was only reachable
    // because id-less entries fell through _applyUpdate's early-return; that path
    // becomes unreachable here for crow_context.
    if (table === "crow_context") {
      try {
        await this._applyCrowContext(op, row, lamport_ts, instance_id);
      } catch (err) {
        console.warn(`[instance-sync] Failed to apply ${op} on crow_context:`, err.message);
      }
      return;
    }

    // Conflict detection gates both updates and deletes — a stale remote delete
    // with a lower lamport_ts must not silently destroy a newer local edit (D6).
    if ((op === "update" || op === "delete") && row.id !== undefined) {
      const conflict = await this._checkConflict(table, row.id, lamport_ts, instance_id, row, op);
      if (conflict === "skip") return; // Local version wins
    }

    // Apply the change
    try {
      switch (op) {
        case "insert":
          await this._applyInsert(table, row, lamport_ts, instance_id);
          break;
        case "update":
          await this._applyUpdate(table, row, lamport_ts, instance_id);
          break;
        case "delete":
          await this._applyDelete(table, row);
          break;
      }
    } catch (err) {
      console.warn(`[instance-sync] Failed to apply ${op} on ${table}:`, err.message);
    }

    // Broadcast a messages:changed event when a synced-in message row
    // lands locally. Without this, a paired Crow receiving a peer
    // message via Nostr forwards the row via InstanceSync but our
    // local onMessage / createNotification paths never fire, so
    // badges wouldn't live-update. The 5-min fallback poll would
    // eventually catch up, but this closes the gap for cross-instance
    // traffic. Errors are swallowed — the row is already applied.
    if (op === "insert" && table === "messages" && row?.contact_id != null) {
      try {
        const { rows } = await this.db.execute({
          sql: `SELECT COUNT(*) AS unread FROM messages
                WHERE contact_id = ? AND is_read = 0 AND direction = 'received'`,
          args: [row.contact_id],
        });
        const unread = Number(rows?.[0]?.unread ?? 0);
        bus.emit("messages:changed", { contactId: row.contact_id, unread });
      } catch {}
    }
  }

  /**
   * Apply a dashboard_settings mutation. Only the global row is synced;
   * per-instance overrides live in dashboard_settings_overrides.
   * Last-write-wins by lamport_ts.
   */
  async _applyDashboardSetting(op, row, lamportTs) {
    if (!row || !row.key) return;

    const { rows: existing } = await this.db.execute({
      sql: `SELECT value, lamport_ts FROM dashboard_settings WHERE key = ?`,
      args: [row.key],
    });
    const localTs = existing[0]?.lamport_ts || 0;
    if (lamportTs < localTs) return;
    if (lamportTs === localTs && existing[0]?.value === row.value) return;
    // Tie + different value falls through: incoming wins silently (no conflict
    // row — operator-only settings rarely diverge; asymmetric with _checkConflict
    // by design, pre-existing behavior).

    if (op === "delete") {
      await this.db.execute({
        sql: `DELETE FROM dashboard_settings WHERE key = ?`,
        args: [row.key],
      });
      return;
    }

    await this.db.execute({
      sql: `INSERT INTO dashboard_settings (key, value, updated_at, lamport_ts)
            VALUES (?, ?, datetime('now'), ?)
            ON CONFLICT(key) DO UPDATE SET
              value = excluded.value, updated_at = datetime('now'), lamport_ts = excluded.lamport_ts`,
      args: [row.key, row.value ?? "", lamportTs],
    });
  }

  /**
   * Apply a crow_context mutation (insert / update / delete).
   *
   * crow_context is keyed by composite (section_key, device_id, project_id) with
   * four NULL-aware partial unique indexes.  Standard _applyUpdate/_applyDelete early-
   * return when row.id is undefined, so we route all ops here instead.
   *
   * Conflict rules follow _checkConflict (W4-1 standard):
   *   incomingTs > localTs  → apply
   *   incomingTs <= localTs, data equal → silent skip
   *   incomingTs <= localTs, data differ (including tie) → conflict, local kept
   *
   * Upsert-on-missing applies to 'update' too (resurrection-over-loss, spec §3 C4):
   * a re-delivered pre-delete update recreates the section rather than silently
   * losing the edit.
   *
   * @param {"insert"|"update"|"delete"} op
   * @param {object} row — wire row (no id, no lamport_ts)
   * @param {number} lamportTs — incoming Lamport timestamp
   * @param {string} instanceId — origin instance id
   */
  async _applyCrowContext(op, row, lamportTs, instanceId) {
    if (!row || row.section_key == null) {
      console.warn("[instance-sync] _applyCrowContext: missing section_key — skipping");
      return;
    }

    const sk = row.section_key;
    const devId = row.device_id ?? null;
    const projId = row.project_id ?? null;

    // Key-filter the incoming row: intersect with live PRAGMA columns; drop
    // id, lamport_ts, instance_id, and updated_at so an unexpected wire key
    // doesn't throw.  Cached per-process on the first crow_context apply.
    if (!this._crowContextCols) {
      try {
        const { rows: pragma } = await this.db.execute({
          sql: "PRAGMA table_info(crow_context)",
          args: [],
        });
        this._crowContextCols = new Set(pragma.map((r) => r.name));
      } catch {
        this._crowContextCols = null;
      }
    }
    const ALWAYS_DROP = new Set(["id", "lamport_ts", "instance_id", "updated_at"]);
    let filteredRow = {};
    for (const [k, v] of Object.entries(row)) {
      if (ALWAYS_DROP.has(k)) continue;
      if (this._crowContextCols && !this._crowContextCols.has(k)) continue;
      filteredRow[k] = v;
    }

    // Read local row
    const { rows: localRows } = await this.db.execute({
      sql: "SELECT * FROM crow_context WHERE section_key = ? AND device_id IS ? AND project_id IS ?",
      args: [sk, devId, projId],
    });
    const localRow = localRows[0] ?? null;
    const localTs = localRow?.lamport_ts || 0;

    // ── delete ────────────────────────────────────────────────────────────────
    if (op === "delete") {
      if (!localRow) return; // Nothing to delete
      if (lamportTs > localTs) {
        await this.db.execute({
          sql: "DELETE FROM crow_context WHERE section_key = ? AND device_id IS ? AND project_id IS ?",
          args: [sk, devId, projId],
        });
        try {
          const { invalidateContextCache } = await import("../memory/crow-context.js");
          invalidateContextCache();
        } catch {}
        return;
      }
      // stale delete → conflict, local kept
      const rowIdJson = JSON.stringify({ section_key: sk, device_id: devId, project_id: projId });
      try {
        await this._insertConflictRow(
          "crow_context", rowIdJson,
          localRow.instance_id || this.localInstanceId, instanceId,
          localTs, lamportTs,
          JSON.stringify(localRow), JSON.stringify(filteredRow),
          "delete",
        );
        await this._notifyConflict();
      } catch (err) {
        console.warn(`[instance-sync] crow_context conflict LOGGING failed (local data preserved):`, err.message);
      }
      return;
    }

    // ── insert / update ───────────────────────────────────────────────────────
    if (!localRow) {
      // No local row: upsert (covers both insert and update ops — resurrection
      // of a deleted section is preferable to silently losing a catch-up edit).
      const required = ["section_title", "content"];
      const missing = required.filter((k) => filteredRow[k] == null);
      if (missing.length > 0) {
        console.warn(`[instance-sync] _applyCrowContext: upsert skipped — NOT NULL column(s) absent (${missing.join(", ")}); old-sender partial entry`);
        return;
      }
      // Include composite key columns that may not be in filteredRow already
      const insertFields = { ...filteredRow, section_key: sk, device_id: devId, project_id: projId, lamport_ts: lamportTs, updated_at: "datetime('now')" };
      // Build the INSERT without updated_at as a bind param (it's an SQL expression)
      const colsForInsert = Object.keys(insertFields).filter((k) => k !== "updated_at");
      const placeholders = colsForInsert.map(() => "?").join(", ");
      const values = colsForInsert.map((k) => insertFields[k]);
      await this.db.execute({
        sql: `INSERT INTO crow_context (${colsForInsert.join(", ")}, updated_at)
              VALUES (${placeholders}, datetime('now'))`,
        args: values,
      });
      try {
        const { invalidateContextCache } = await import("../memory/crow-context.js");
        invalidateContextCache();
      } catch {}
      return;
    }

    // Local row exists
    if (lamportTs > localTs) {
      // Newer incoming: UPDATE present filtered keys + lamport_ts + updated_at
      const updateKeys = Object.keys(filteredRow).filter((k) => !["section_key", "device_id", "project_id"].includes(k));
      if (updateKeys.length > 0) {
        const setClauses = [...updateKeys.map((k) => `${k} = ?`), "lamport_ts = ?", "updated_at = datetime('now')"].join(", ");
        const vals = [...updateKeys.map((k) => filteredRow[k] ?? null), lamportTs];
        await this.db.execute({
          sql: `UPDATE crow_context SET ${setClauses} WHERE section_key = ? AND device_id IS ? AND project_id IS ?`,
          args: [...vals, sk, devId, projId],
        });
      } else {
        // Nothing to update besides timestamps
        await this.db.execute({
          sql: `UPDATE crow_context SET lamport_ts = ?, updated_at = datetime('now') WHERE section_key = ? AND device_id IS ? AND project_id IS ?`,
          args: [lamportTs, sk, devId, projId],
        });
      }
      try {
        const { invalidateContextCache } = await import("../memory/crow-context.js");
        invalidateContextCache();
      } catch {}
      return;
    }

    // incomingTs <= localTs: check equivalence
    if (rowsEquivalent(localRow, filteredRow)) {
      return; // Re-delivery noise — silent skip
    }

    // Real conflict (includes tie + different data): local kept
    const rowIdJson = JSON.stringify({ section_key: sk, device_id: devId, project_id: projId });
    try {
      await this._insertConflictRow(
        "crow_context", rowIdJson,
        localRow.instance_id || this.localInstanceId, instanceId,
        localTs, lamportTs,
        JSON.stringify(localRow), JSON.stringify(filteredRow),
        op || "update",
      );
      await this._notifyConflict();
    } catch (err) {
      console.warn(`[instance-sync] crow_context conflict LOGGING failed (local data preserved):`, err.message);
    }
  }

  /**
   * Check if applying an update or delete would conflict with a local version.
   *
   * Equivalence check (for updates): apply the table's OUTBOUND_TRANSFORMS to
   * the local row before comparing, so a locally-assigned transformed column
   * (e.g. research_notes.project_id) never manufactures a false conflict.
   *
   * Delete path: equivalence check is skipped — a delete vs a live row is never
   * equivalent. Low incomingTs + local row → conflict op='delete'.
   *
   * NO same-author heuristic: row.instance_id is the ORIGIN instance, not the
   * last editor. Treating same-origin as "safe stale re-order" would silently
   * drop real concurrent edits of rows both instances received from one origin.
   *
   * @param {string} op - 'update' or 'delete' (caller gates before calling)
   * @returns {Promise<"apply"|"skip">}
   */
  async _checkConflict(table, rowId, incomingTs, incomingInstanceId, incomingRow, op) {
    try {
      const { rows } = await this.db.execute({
        sql: `SELECT * FROM ${table} WHERE id = ?`,
        args: [rowId],
      });

      if (rows.length === 0) return "apply"; // No local version — safe to apply

      const localRow = rows[0];
      const localTs = localRow.lamport_ts || 0;

      if (incomingTs > localTs) {
        return "apply"; // Incoming is newer
      }

      // incomingTs <= localTs: potential conflict.
      // For updates: run equivalence check to filter re-delivery noise.
      // For deletes: skip equivalence — a delete vs live row is never equivalent.
      if (op !== "delete") {
        // Apply OUTBOUND_TRANSFORMS to the local row before comparing so that a
        // locally-assigned transformed column (e.g. project_id on research_notes)
        // doesn't manufacture a false conflict for an otherwise-identical row.
        const transform = OUTBOUND_TRANSFORMS[table];
        const localForCompare = transform ? transform({ ...localRow }) : localRow;
        if (rowsEquivalent(localForCompare, incomingRow)) {
          return "skip"; // Re-delivery noise — no conflict row needed
        }
      }

      // Real conflict: log both versions and notify the operator.
      // The verdict is already determined here (local is newer and the data
      // differs) — a failure to LOG must not flip it to "apply" and overwrite
      // newer local data, so logging gets its own guard and we skip regardless.
      try {
        await this._insertConflictRow(
          table, String(rowId),
          localRow.instance_id || this.localInstanceId, incomingInstanceId,
          localTs, incomingTs,
          JSON.stringify(localRow), JSON.stringify(incomingRow),
          op || "update",
        );
        await this._notifyConflict();
      } catch (err) {
        console.warn(`[instance-sync] Conflict LOGGING failed for ${table}:${rowId} (local data still preserved):`, err.message);
      }

      return "skip"; // Local version wins
    } catch (err) {
      console.warn(`[instance-sync] Conflict check failed for ${table}:${rowId}:`, err.message);
    }

    return "apply"; // Default to applying on error
  }

  /**
   * Insert a sync_conflicts row, degrading gracefully when the `op` column
   * doesn't exist yet (code pulled + gateway restarted before init-db ran —
   * the gateway-boot migration also closes this, but a non-gateway host or a
   * mid-boot race must not lose the conflict trace, and in _checkConflict a
   * thrown INSERT must never cascade into applying a stale row).
   */
  async _insertConflictRow(tableName, rowId, winInst, loseInst, winTs, loseTs, winData, loseData, conflictOp) {
    const legacyCols = `table_name, row_id, winning_instance_id, losing_instance_id,
                 winning_lamport_ts, losing_lamport_ts, winning_data, losing_data`;
    const legacyArgs = [tableName, rowId, winInst, loseInst, winTs, loseTs, winData, loseData];
    try {
      await this.db.execute({
        sql: `INSERT INTO sync_conflicts (${legacyCols}, op) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
        args: [...legacyArgs, conflictOp],
      });
    } catch (err) {
      if (!/no column named op/i.test(err.message || "")) throw err;
      await this.db.execute({
        sql: `INSERT INTO sync_conflicts (${legacyCols}) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
        args: legacyArgs,
      });
    }
  }

  /**
   * Fire an operator notification when a real conflict is logged.
   * Deduped: if an unread+undismissed instance-sync notification already
   * exists, skip creating another one. The SELECT-then-INSERT is TOCTOU-racy
   * across peers (per-peer locks don't serialize across different peers), so a
   * duplicate may slip through — that's acceptable (harmless extra bell item).
   * Never throws into the apply loop.
   */
  async _notifyConflict() {
    try {
      const { rows: existing } = await this.db.execute({
        sql: `SELECT id FROM notifications
              WHERE source = 'instance-sync' AND is_read = 0 AND is_dismissed = 0
              LIMIT 1`,
        args: [],
      });
      if (existing.length > 0) return; // Standing notification already present

      // Lazy dynamic import to avoid loading gateway push modules in
      // non-gateway contexts (same pattern as crow-context.js import above).
      const { createNotification } = await import("../shared/notifications.js");
      await createNotification(this.db, {
        type: "system",
        source: "instance-sync",
        priority: "high",
        title: "Sync conflict recorded",
        body: "A sync conflict was recorded — one version of an item was kept, the other saved for review.",
        action_url: "/dashboard/settings?section=sync-conflicts",
      });
    } catch (err) {
      console.warn("[instance-sync] Failed to send conflict notification:", err.message);
    }
  }

  /**
   * Apply an insert operation from a remote instance.
   * On INSERT OR IGNORE conflict (rowsAffected === 0): fetch the local row.
   * - row.id == null → warn only (binding undefined to better-sqlite3 throws)
   * - No local row at that id → secondary UNIQUE/NOT NULL/CHECK collision, not
   *   an id collision; warn-log only, no conflict row (the dual-path messages
   *   Nostr+sync delivery is the live example)
   * - Local row equivalent → benign re-delivery, done
   * - Local row differs → log conflict op='insert' + notify (D7 surfacing)
   */
  async _applyInsert(table, row, lamportTs, instanceId) {
    const cols = Object.keys(row);
    // Add sync metadata
    if (!cols.includes("lamport_ts")) cols.push("lamport_ts");
    if (!cols.includes("instance_id") && table === "memories") cols.push("instance_id");

    const values = cols.map(c => {
      if (c === "lamport_ts") return lamportTs;
      if (c === "instance_id" && row[c] === undefined) return instanceId;
      return row[c] ?? null;
    });

    const placeholders = cols.map(() => "?").join(", ");
    const colNames = cols.join(", ");

    const result = await this.db.execute({
      sql: `INSERT OR IGNORE INTO ${table} (${colNames}) VALUES (${placeholders})`,
      args: values,
    });

    if (result.rowsAffected === 0) {
      // INSERT OR IGNORE skipped — figure out why.
      if (row.id == null) {
        // Better-sqlite3 throws on undefined binds; null id is also not useful.
        console.warn(`[instance-sync] _applyInsert: null/undefined id on ${table} — skipping`);
        return;
      }

      const { rows: localRows } = await this.db.execute({
        sql: `SELECT * FROM ${table} WHERE id = ?`,
        args: [row.id],
      });

      if (localRows.length === 0) {
        // No row at this id — secondary UNIQUE/NOT NULL/CHECK collision.
        // Not an id collision; log and move on without a conflict row.
        console.warn(`[instance-sync] _applyInsert: INSERT OR IGNORE on ${table} id=${row.id} skipped (secondary constraint); no conflict logged`);
        return;
      }

      const localRow = localRows[0];
      // Apply transform to the local row before equivalence comparison (same
      // logic as _checkConflict) so a transformed column doesn't trigger
      // a false insert-conflict for an otherwise-identical row.
      const transform = OUTBOUND_TRANSFORMS[table];
      const localForCompare = transform ? transform({ ...localRow }) : localRow;
      if (rowsEquivalent(localForCompare, row)) {
        return; // Benign re-delivery
      }

      // Id collision with different data — surface it (D7 minimal fix).
      // The incoming insert is still not applied; the trace is preserved.
      await this._insertConflictRow(
        table, String(row.id),
        localRow.instance_id || this.localInstanceId, instanceId,
        localRow.lamport_ts || 0, lamportTs,
        JSON.stringify(localRow), JSON.stringify(row),
        "insert",
      );

      await this._notifyConflict();
    }
  }

  /**
   * Apply an update operation from a remote instance.
   */
  async _applyUpdate(table, row, lamportTs) {
    if (row.id === undefined) return;

    const updates = [];
    const params = [];

    for (const [key, value] of Object.entries(row)) {
      if (key === "id") continue;
      updates.push(`${key} = ?`);
      params.push(value ?? null);
    }

    // Always update lamport_ts
    updates.push("lamport_ts = ?");
    params.push(lamportTs);

    params.push(row.id);

    await this.db.execute({
      sql: `UPDATE ${table} SET ${updates.join(", ")} WHERE id = ?`,
      args: params,
    });
  }

  /**
   * Apply a delete operation from a remote instance.
   */
  async _applyDelete(table, row) {
    if (row.id === undefined) return;

    await this.db.execute({
      sql: `DELETE FROM ${table} WHERE id = ?`,
      args: [row.id],
    });
  }

  /**
   * Get the last applied sequence number for a remote instance's feed.
   */
  async _getLastAppliedSeq(remoteInstanceId) {
    try {
      const { rows } = await this.db.execute({
        sql: "SELECT last_applied_seq_per_peer FROM sync_state WHERE instance_id = ?",
        args: [this.localInstanceId],
      });

      if (rows.length > 0 && rows[0].last_applied_seq_per_peer) {
        const seqs = JSON.parse(rows[0].last_applied_seq_per_peer);
        return seqs[remoteInstanceId] || 0;
      }
    } catch {}

    return 0;
  }

  /**
   * Atomically update the last applied sequence number for a remote peer.
   *
   * Uses json_set on the existing blob — a single atomic UPDATE that avoids
   * the SELECT-then-UPDATE race (D3) where two peers checkpointing concurrently
   * could lose one of the two writes. json_set accepts the path as a bound
   * parameter; instance ids are UUIDs so they contain no `"` — but we
   * defensively reject ids that would corrupt the JSON path.
   *
   * Semantics unchanged: `seq` is the NEXT unprocessed sequence number
   * (i.e. the last successfully attempted seq + 1). getSyncStatus and
   * _getLastAppliedSeq consume the same blob.
   */
  async _setLastAppliedSeq(remoteInstanceId, seq) {
    // Guard: a `"` in the id would break the json_set path literal.
    if (remoteInstanceId.includes('"')) {
      console.warn(`[instance-sync] _setLastAppliedSeq: skipping id with quote char: ${remoteInstanceId}`);
      return;
    }
    try {
      await this._ensureCounter();
      await this.db.execute({
        sql: `UPDATE sync_state
              SET last_applied_seq_per_peer = json_set(COALESCE(last_applied_seq_per_peer, '{}'), ?, CAST(? AS INTEGER)),
                  updated_at = datetime('now')
              WHERE instance_id = ?`,
        args: [`$."${remoteInstanceId}"`, seq, this.localInstanceId],
      });
    } catch (err) {
      console.warn(`[instance-sync] Failed to update checkpoint for ${remoteInstanceId}:`, err.message);
    }
  }

  /**
   * Get sync status summary for all connected instances.
   */
  async getSyncStatus() {
    const status = [];

    for (const [instanceId, feed] of this.outFeeds) {
      const inFeed = this.inFeeds.get(instanceId);
      const lastSeq = await this._getLastAppliedSeq(instanceId);

      status.push({
        instanceId,
        outFeedLength: feed.length,
        inFeedLength: inFeed ? inFeed.length : 0,
        lastAppliedSeq: lastSeq,
        pendingEntries: inFeed ? Math.max(0, inFeed.length - lastSeq) : 0,
      });
    }

    return status;
  }

  /**
   * Close feeds for a single remote instance and remove them from the Maps.
   *
   * Serialized through the _initLocks tail so close cannot interleave with a
   * concurrent initInstance call for the same remoteInstanceId. Hypercore close
   * is safe and on-disk storage persists; the instance lazily re-inits on
   * un-revoke (boot.js eagerInitPairedPeers / tailnet-sync paths gate on status
   * and will reopen when the instance is un-revoked).
   */
  async closeInstanceFeeds(remoteInstanceId) {
    const prior = this._initLocks.get(remoteInstanceId) || Promise.resolve();
    const next = prior
      .catch(() => {})
      .then(() => this._closeInstanceFeedsInner(remoteInstanceId));
    this._initLocks.set(remoteInstanceId, next);
    try {
      return await next;
    } finally {
      if (this._initLocks.get(remoteInstanceId) === next) {
        this._initLocks.delete(remoteInstanceId);
      }
    }
  }

  async _closeInstanceFeedsInner(remoteInstanceId) {
    const outFeed = this.outFeeds.get(remoteInstanceId);
    if (outFeed) {
      try { await outFeed.close(); } catch {}
      this.outFeeds.delete(remoteInstanceId);
    }
    const inFeed = this.inFeeds.get(remoteInstanceId);
    if (inFeed) {
      try { await inFeed.close(); } catch {}
      this.inFeeds.delete(remoteInstanceId);
    }
  }

  /**
   * Close all feeds.
   */
  async close() {
    for (const feed of this.outFeeds.values()) {
      try { await feed.close(); } catch {}
    }
    for (const feed of this.inFeeds.values()) {
      try { await feed.close(); } catch {}
    }
    this.outFeeds.clear();
    this.inFeeds.clear();
    this.started = false;
  }
}
