// 同步核心：push（LWW 仲裁）+ pull（seq 游标增量）
import { getDb, SYNC_TABLES } from './db.mjs';

function parseHlc(h) {
  const m = /^([0-9]+):([0-9]+):(.+)$/.exec(String(h || ''));
  if (!m) return null;
  return { ms: Number(m[1]), cnt: Number(m[2]), dev: m[3] };
}

// a 是否严格新于 b：比较 (ms, cnt, dev)
function isNewer(a, b) {
  if (a.ms !== b.ms) return a.ms > b.ms;
  if (a.cnt !== b.cnt) return a.cnt > b.cnt;
  return a.dev > b.dev;
}

export function push(user_id, changes) {
  const d = getDb();
  const stats = { accepted: 0, stale: 0, invalid: 0 };
  if (!Array.isArray(changes)) throw new Error('changes 必须为数组');
  if (changes.length > 5000) throw new Error('单次推送不得超过 5000 条');

  for (const c of changes) {
    const table = c.table;
    const row_uuid = String(c.row_uuid || '');
    const op = c.op;
    const parsed = parseHlc(c.hlc);
    if (!SYNC_TABLES.has(table) || !parsed || !['upsert', 'delete'].includes(op)) {
      stats.invalid++;
      continue;
    }
    if (op === 'upsert' && (c.data == null || typeof c.data !== 'object')) {
      stats.invalid++;
      continue;
    }
    const dataJson = op === 'upsert' ? JSON.stringify(c.data) : null;

    const existing = d.prepare(
      'SELECT ms, cnt, dev FROM current WHERE user_id=? AND table_name=? AND row_uuid=?'
    ).get(user_id, table, row_uuid);

    if (existing) {
      const e = { ms: existing.ms, cnt: existing.cnt, dev: existing.dev };
      if (parsed.ms === e.ms && parsed.cnt === e.cnt && parsed.dev === e.dev) {
        stats.accepted++; // 同一变更重放，幂等
        continue;
      }
      if (!isNewer(parsed, e)) { stats.stale++; continue; }
      d.prepare(`UPDATE current SET op=?, hlc=?, ms=?, cnt=?, dev=?, data=?,
        updated_seq=(SELECT COALESCE(MAX(seq),0)+1 FROM changes) WHERE user_id=? AND table_name=? AND row_uuid=?`)
        .run(op, c.hlc, parsed.ms, parsed.cnt, parsed.dev, dataJson, user_id, table, row_uuid);
      d.prepare('INSERT INTO changes (user_id, table_name, row_uuid, op, hlc, data) VALUES (?,?,?,?,?,?)')
        .run(user_id, table, row_uuid, op, c.hlc, dataJson);
      stats.accepted++;
    } else {
      d.prepare(`INSERT INTO current (user_id, table_name, row_uuid, op, hlc, ms, cnt, dev, data,
        updated_seq) VALUES (?,?,?,?,?,?,?,?,?,(SELECT COALESCE(MAX(seq),0)+1 FROM changes))`)
        .run(user_id, table, row_uuid, op, c.hlc, parsed.ms, parsed.cnt, parsed.dev, dataJson);
      d.prepare('INSERT INTO changes (user_id, table_name, row_uuid, op, hlc, data) VALUES (?,?,?,?,?,?)')
        .run(user_id, table, row_uuid, op, c.hlc, dataJson);
      stats.accepted++;
    }
  }
  return stats;
}

export function pull(user_id, cursorRaw, limitRaw) {
  const d = getDb();
  const cursor = Math.max(0, Number(cursorRaw) || 0);
  const limit = Math.min(2000, Math.max(1, Number(limitRaw) || 1000));
  const rows = d.prepare(
    'SELECT updated_seq AS seq, table_name, row_uuid, op, hlc, data FROM current WHERE user_id=? AND updated_seq > ? ORDER BY seq LIMIT ?'
  ).all(user_id, cursor, limit);
  const changes = rows.map(r => ({
    seq: r.seq,
    table: r.table_name,
    row_uuid: r.row_uuid,
    op: r.op,
    hlc: r.hlc,
    data: r.data ? JSON.parse(r.data) : null
  }));
  const maxSeqRow = d.prepare('SELECT COALESCE(MAX(updated_seq),0) AS m FROM current WHERE user_id=?').get(user_id);
  const hasMore = rows.length === limit;
  return {
    changes,
    cursor: hasMore ? rows[rows.length - 1].seq : Math.max(cursor, maxSeqRow.m),
    has_more: hasMore
  };
}