'use strict';

const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const { getSyncContext, requireLocalIdentity } = require('./sync.context');
const { assertEventInput, sanitizePayload } = require('./sync.schema');
const { assertTenantScope } = require('./sync.conflicts');

const DEFAULT_MAX_RETRIES = parseInt(process.env.ARKHON_SYNC_MAX_RETRIES || process.env.SYNC_MAX_RETRIES || '10', 10);
const DEFAULT_LEASE_SECONDS = parseInt(process.env.ARKHON_SYNC_LEASE_SECONDS || '300', 10);
const FILE_DRIVER_ALLOWED = process.env.NODE_ENV !== 'production' && (process.env.NODE_ENV === 'test' || process.env.ARKHON_SYNC_STORE_DRIVER === 'file-test');
let dbModule = null;

function db() {
  if (!dbModule) dbModule = require('../../db');
  return dbModule;
}

function driver() {
  return FILE_DRIVER_ALLOWED ? 'file' : 'sqlserver';
}

function getStorePath() {
  return process.env.ARKHON_SYNC_STORE_PATH || path.join(__dirname, '..', '..', '..', 'config', 'sync-state.local.json');
}

function emptyStore() {
  return { version: 2, events: [], batches: [], commands: [], state: { last_checkpoint: null, last_sync_at: null }, audit: [] };
}

function readFileStore(filePath = getStorePath()) {
  if (!fs.existsSync(filePath)) return emptyStore();
  return { ...emptyStore(), ...JSON.parse(fs.readFileSync(filePath, 'utf8')) };
}

function writeFileStore(store, filePath = getStorePath()) {
  fs.mkdirSync(path.dirname(filePath), { recursive: true });
  const tmp = `${filePath}.${process.pid}.tmp`;
  fs.writeFileSync(tmp, JSON.stringify(store, null, 2), 'utf8');
  fs.renameSync(tmp, filePath);
}

async function updateFile(mutator, filePath = getStorePath()) {
  const store = readFileStore(filePath);
  const next = mutator(store) || store;
  writeFileStore(next, filePath);
  return next;
}

async function ensureSchema(pool = null) {
  if (driver() === 'file') return;
  const { getPool } = db();
  const dbPool = pool || await getPool();
  await dbPool.request().batch(`
IF OBJECT_ID('dbo.ArkhonSyncState','U') IS NULL
CREATE TABLE dbo.ArkhonSyncState (
  [key] NVARCHAR(100) NOT NULL PRIMARY KEY,
  [value] NVARCHAR(MAX) NULL,
  updated_at DATETIME2 NOT NULL DEFAULT SYSUTCDATETIME()
);
IF OBJECT_ID('dbo.ArkhonSyncEvents','U') IS NULL
CREATE TABLE dbo.ArkhonSyncEvents (
  event_id NVARCHAR(100) NOT NULL PRIMARY KEY,
  tenant_id NVARCHAR(100) NOT NULL,
  branch_id NVARCHAR(100) NOT NULL,
  connector_id NVARCHAR(100) NOT NULL,
  entity_type NVARCHAR(160) NOT NULL,
  entity_id NVARCHAR(200) NOT NULL,
  operation NVARCHAR(30) NOT NULL,
  payload_json NVARCHAR(MAX) NOT NULL,
  payload_hash NVARCHAR(128) NOT NULL,
  idempotency_key NVARCHAR(200) NOT NULL UNIQUE,
  status NVARCHAR(30) NOT NULL,
  retry_count INT NOT NULL DEFAULT 0,
  next_retry_at DATETIME2 NOT NULL DEFAULT SYSUTCDATETIME(),
  batch_id NVARCHAR(100) NULL,
  lease_id NVARCHAR(100) NULL,
  lease_until DATETIME2 NULL,
  leased_at DATETIME2 NULL,
  last_error NVARCHAR(1000) NULL,
  created_at DATETIME2 NOT NULL DEFAULT SYSUTCDATETIME(),
  updated_at DATETIME2 NOT NULL DEFAULT SYSUTCDATETIME(),
  synced_at DATETIME2 NULL
);
IF OBJECT_ID('dbo.ArkhonSyncBatches','U') IS NULL
CREATE TABLE dbo.ArkhonSyncBatches (
  batch_id NVARCHAR(100) NOT NULL PRIMARY KEY,
  event_ids NVARCHAR(MAX) NOT NULL,
  lease_id NVARCHAR(100) NULL,
  status NVARCHAR(30) NOT NULL,
  created_at DATETIME2 NOT NULL DEFAULT SYSUTCDATETIME(),
  updated_at DATETIME2 NULL
);
IF OBJECT_ID('dbo.ArkhonCloudCommands','U') IS NULL
CREATE TABLE dbo.ArkhonCloudCommands (
  command_id NVARCHAR(100) NOT NULL PRIMARY KEY,
  command_type NVARCHAR(160) NOT NULL,
  idempotency_key NVARCHAR(200) NOT NULL UNIQUE,
  payload_json NVARCHAR(MAX) NOT NULL,
  status NVARCHAR(30) NOT NULL,
  result_json NVARCHAR(MAX) NULL,
  error NVARCHAR(1000) NULL,
  received_at DATETIME2 NOT NULL DEFAULT SYSUTCDATETIME(),
  applying_at DATETIME2 NULL,
  applied_at DATETIME2 NULL,
  updated_at DATETIME2 NULL
);
IF OBJECT_ID('dbo.ArkhonSyncAudit','U') IS NULL
CREATE TABLE dbo.ArkhonSyncAudit (
  id BIGINT IDENTITY(1,1) NOT NULL PRIMARY KEY,
  event_name NVARCHAR(160) NOT NULL,
  entity_id NVARCHAR(200) NULL,
  payload_json NVARCHAR(MAX) NULL,
  created_at DATETIME2 NOT NULL DEFAULT SYSUTCDATETIME()
);
`);
  await dbPool.request().batch(`
IF COL_LENGTH('dbo.ArkhonSyncEvents','lease_id') IS NULL ALTER TABLE dbo.ArkhonSyncEvents ADD lease_id NVARCHAR(100) NULL;
IF COL_LENGTH('dbo.ArkhonSyncEvents','lease_until') IS NULL ALTER TABLE dbo.ArkhonSyncEvents ADD lease_until DATETIME2 NULL;
IF COL_LENGTH('dbo.ArkhonSyncEvents','leased_at') IS NULL ALTER TABLE dbo.ArkhonSyncEvents ADD leased_at DATETIME2 NULL;
IF COL_LENGTH('dbo.ArkhonSyncBatches','lease_id') IS NULL ALTER TABLE dbo.ArkhonSyncBatches ADD lease_id NVARCHAR(100) NULL;
`);
  await migrateLegacyJsonIfNeeded(dbPool);
  await recoverExpiredLeases(dbPool);
}

async function migrateLegacyJsonIfNeeded(db) {
  const legacyPath = getStorePath();
  if (!fs.existsSync(legacyPath)) return;
  const { sql } = dbModule || db();
  const migrated = await db.request().input('key', sql.NVarChar, 'legacy_json_migrated').query('SELECT [value] FROM dbo.ArkhonSyncState WHERE [key]=@key');
  if (migrated.recordset.length) return;
  let legacy;
  try {
    legacy = readFileStore(legacyPath);
  } catch (_) {
    return;
  }
  await withTransaction(async (tx) => {
    for (const event of legacy.events || []) {
      await insertEventSql(event, tx);
    }
    await tx.request()
      .input('key', sql.NVarChar, 'legacy_json_migrated')
      .input('value', sql.NVarChar(sql.MAX), new Date().toISOString())
      .query('MERGE dbo.ArkhonSyncState AS t USING (SELECT @key AS [key], @value AS [value]) AS s ON t.[key]=s.[key] WHEN MATCHED THEN UPDATE SET [value]=s.[value], updated_at=SYSUTCDATETIME() WHEN NOT MATCHED THEN INSERT ([key],[value]) VALUES (s.[key],s.[value]);');
  }, db);
}

async function withTransaction(fn, existingPool = null) {
  const { getPool, sql } = dbModule || require('../../db');
  const pool = existingPool || await getPool();
  const tx = new sql.Transaction(pool);
  await tx.begin();
  try {
    const result = await fn(tx, sql);
    await tx.commit();
    return result;
  } catch (err) {
    await tx.rollback().catch(() => {});
    throw err;
  }
}

async function recoverExpiredLeases(pool = null) {
  if (driver() === 'file') return;
  const { getPool } = db();
  const dbPool = pool || await getPool();
  await dbPool.request().query(`
UPDATE dbo.ArkhonSyncEvents
SET status='failed',
    last_error=COALESCE(last_error, 'lease_expired'),
    next_retry_at=SYSUTCDATETIME(),
    lease_id=NULL,
    lease_until=NULL,
    leased_at=NULL,
    updated_at=SYSUTCDATETIME()
WHERE status='in_flight' AND lease_until IS NOT NULL AND lease_until < SYSUTCDATETIME();
`);
}

function hashPayload(payload) {
  return crypto.createHash('sha256').update(JSON.stringify(payload || {})).digest('hex');
}

function buildIdempotencyKey(context, input, payloadHash) {
  return crypto.createHash('sha256')
    .update([context.connectorId, input.entity_type, input.entity_id, input.operation, payloadHash].join(':'))
    .digest('hex');
}

function fromSqlEvent(row) {
  return {
    event_id: row.event_id,
    tenant_id: row.tenant_id,
    branch_id: row.branch_id,
    connector_id: row.connector_id,
    entity_type: row.entity_type,
    entity_id: row.entity_id,
    operation: row.operation,
    payload_json: JSON.parse(row.payload_json || '{}'),
    payload_hash: row.payload_hash,
    idempotency_key: row.idempotency_key,
    status: row.status,
    retry_count: row.retry_count,
    next_retry_at: row.next_retry_at?.toISOString?.() || row.next_retry_at,
    batch_id: row.batch_id,
    lease_id: row.lease_id,
    lease_until: row.lease_until?.toISOString?.() || row.lease_until,
    leased_at: row.leased_at?.toISOString?.() || row.leased_at,
    last_error: row.last_error,
    created_at: row.created_at?.toISOString?.() || row.created_at,
    updated_at: row.updated_at?.toISOString?.() || row.updated_at,
    synced_at: row.synced_at?.toISOString?.() || row.synced_at,
  };
}

async function insertEventSql(event, db = null) {
  const { getPool, sql } = dbModule || require('../../db');
  const pool = db || await getPool();
  await pool.request()
    .input('event_id', sql.NVarChar, event.event_id)
    .input('tenant_id', sql.NVarChar, event.tenant_id)
    .input('branch_id', sql.NVarChar, event.branch_id)
    .input('connector_id', sql.NVarChar, event.connector_id)
    .input('entity_type', sql.NVarChar, event.entity_type)
    .input('entity_id', sql.NVarChar, String(event.entity_id))
    .input('operation', sql.NVarChar, event.operation)
    .input('payload_json', sql.NVarChar(sql.MAX), JSON.stringify(event.payload_json || {}))
    .input('payload_hash', sql.NVarChar, event.payload_hash)
    .input('idempotency_key', sql.NVarChar, event.idempotency_key)
    .input('status', sql.NVarChar, event.status || 'pending')
    .query(`
IF NOT EXISTS (SELECT 1 FROM dbo.ArkhonSyncEvents WHERE event_id=@event_id OR idempotency_key=@idempotency_key)
INSERT INTO dbo.ArkhonSyncEvents (event_id, tenant_id, branch_id, connector_id, entity_type, entity_id, operation, payload_json, payload_hash, idempotency_key, status)
VALUES (@event_id,@tenant_id,@branch_id,@connector_id,@entity_type,@entity_id,@operation,@payload_json,@payload_hash,@idempotency_key,@status);
`);
}

async function createEvent(input, context = getSyncContext()) {
  requireLocalIdentity(context);
  const data = assertEventInput({ ...input });
  const payload = sanitizePayload(data.payload || {});
  const payloadHash = hashPayload(payload);
  const now = new Date().toISOString();
  const event = {
    event_id: data.event_id || crypto.randomUUID(),
    tenant_id: context.tenantId,
    branch_id: context.branchId,
    connector_id: context.connectorId,
    entity_type: data.entity_type,
    entity_id: String(data.entity_id),
    operation: data.operation,
    payload_json: payload,
    payload_hash: payloadHash,
    idempotency_key: data.idempotency_key || buildIdempotencyKey(context, data, payloadHash),
    status: 'pending',
    retry_count: 0,
    next_retry_at: now,
    last_error: null,
    created_at: now,
    updated_at: now,
    synced_at: null,
  };
  assertTenantScope(event, context);
  if (driver() === 'file') {
    await updateFile((store) => {
      if (!store.events.some((existing) => existing.idempotency_key === event.idempotency_key || existing.event_id === event.event_id)) store.events.push(event);
      return store;
    });
    return event;
  }
  await ensureSchema();
  await withTransaction(async (tx) => insertEventSql(event, tx));
  return event;
}

async function listEvents({ status } = {}) {
  if (driver() === 'file') {
    const events = readFileStore().events;
    return status ? events.filter((event) => event.status === status) : events;
  }
  await ensureSchema();
  const { getPool, sql } = db();
  const pool = await getPool();
  const request = pool.request();
  const where = status ? 'WHERE status=@status' : '';
  if (status) request.input('status', sql.NVarChar, status);
  const result = await request.query(`SELECT * FROM dbo.ArkhonSyncEvents ${where} ORDER BY created_at ASC`);
  return result.recordset.map(fromSqlEvent);
}

async function getQueueStats() {
  if (driver() === 'file') {
    const store = readFileStore();
    return store.events.reduce((acc, event) => {
      acc[event.status] = (acc[event.status] || 0) + 1;
      return acc;
    }, { total: store.events.length, last_checkpoint: store.state.last_checkpoint, last_sync_at: store.state.last_sync_at });
  }
  await ensureSchema();
  const { getPool } = db();
  const pool = await getPool();
  const counts = await pool.request().query('SELECT status, COUNT(*) AS count FROM dbo.ArkhonSyncEvents GROUP BY status');
  const state = await pool.request().query("SELECT [key], [value] FROM dbo.ArkhonSyncState WHERE [key] IN ('last_checkpoint','last_sync_at')");
  const out = { total: 0, last_checkpoint: null, last_sync_at: null };
  for (const row of counts.recordset) {
    out[row.status] = Number(row.count);
    out.total += Number(row.count);
  }
  for (const row of state.recordset) out[row.key] = row.value;
  return out;
}

async function dueEvents(limit = 100, now = new Date()) {
  if (driver() === 'file') {
    return readFileStore().events.filter((event) => (['pending', 'failed'].includes(event.status) && new Date(event.next_retry_at || 0) <= now) || (event.status === 'in_flight' && event.lease_until && new Date(event.lease_until) <= now)).slice(0, limit);
  }
  await ensureSchema();
  const { getPool, sql } = db();
  const pool = await getPool();
  const result = await pool.request()
    .input('limit', sql.Int, limit)
    .input('now', sql.DateTime2, now)
    .query("SELECT TOP (@limit) * FROM dbo.ArkhonSyncEvents WHERE (status IN ('pending','failed') AND next_retry_at <= @now) OR (status='in_flight' AND lease_until IS NOT NULL AND lease_until <= @now) ORDER BY created_at ASC");
  return result.recordset.map(fromSqlEvent);
}

async function markSyncing(eventIds, batchId, options = {}) {
  if (!Array.isArray(eventIds) || eventIds.length === 0) throw new Error('markSyncing requiere eventos');
  if (!batchId) throw new Error('batch_id requerido para markSyncing');
  const leaseId = options.leaseId || crypto.randomUUID();
  const leaseSeconds = options.leaseSeconds || DEFAULT_LEASE_SECONDS;
  if (driver() === 'file') {
    const ids = new Set(eventIds);
    await updateFile((store) => {
      const eligible = store.events.filter((event) => ids.has(event.event_id) && (['pending', 'failed'].includes(event.status) || (event.status === 'in_flight' && event.lease_until && new Date(event.lease_until) <= new Date())));
      if (eligible.length !== ids.size) throw new Error('No se pudo leasear todos los eventos solicitados');
      const now = new Date().toISOString();
      const leaseUntil = new Date(Date.now() + leaseSeconds * 1000).toISOString();
      store.events.forEach((event) => { if (eligible.some((item) => item.event_id === event.event_id)) { event.status = 'in_flight'; event.batch_id = batchId; event.lease_id = leaseId; event.leased_at = now; event.lease_until = leaseUntil; event.updated_at = now; } });
      store.batches.push({ batch_id: batchId, event_ids: eventIds, lease_id: leaseId, status: 'in_flight', created_at: now });
      return store;
    });
    return { lease_id: leaseId, lease_until: new Date(Date.now() + leaseSeconds * 1000).toISOString() };
  }
  await ensureSchema();
  const { sql } = db();
  await withTransaction(async (tx) => {
    for (const eventId of eventIds) {
      const updated = await tx.request()
        .input('event_id', sql.NVarChar, eventId)
        .input('batch_id', sql.NVarChar, batchId)
        .input('lease_id', sql.NVarChar, leaseId)
        .input('lease_seconds', sql.Int, leaseSeconds)
        .query("UPDATE dbo.ArkhonSyncEvents SET status='in_flight', batch_id=@batch_id, lease_id=@lease_id, leased_at=SYSUTCDATETIME(), lease_until=DATEADD(second,@lease_seconds,SYSUTCDATETIME()), updated_at=SYSUTCDATETIME() WHERE event_id=@event_id AND (status IN ('pending','failed') OR (status='in_flight' AND lease_until <= SYSUTCDATETIME()))");
      if (!updated.rowsAffected?.[0]) throw new Error(`No se pudo leasear evento ${eventId}`);
    }
    await tx.request()
      .input('batch_id', sql.NVarChar, batchId)
      .input('event_ids', sql.NVarChar(sql.MAX), JSON.stringify(eventIds))
      .input('lease_id', sql.NVarChar, leaseId)
      .query("INSERT INTO dbo.ArkhonSyncBatches (batch_id, event_ids, lease_id, status) VALUES (@batch_id,@event_ids,@lease_id,'in_flight')");
  });
  return { lease_id: leaseId };
}

async function applyAck(ack = {}) {
  if (driver() === 'file') {
    const accepted = new Set(ack.accepted_event_ids || []);
    const rejected = new Set(ack.rejected_event_ids || []);
    await updateFile((store) => {
      const now = new Date().toISOString();
      store.events.forEach((event) => {
        if (ack.batch_id && event.batch_id !== ack.batch_id) return;
        if (ack.lease_id && event.lease_id !== ack.lease_id) return;
        if (accepted.has(event.event_id)) Object.assign(event, { status: 'synced', synced_at: now, updated_at: now, last_error: null, lease_id: null, lease_until: null });
        if (rejected.has(event.event_id)) Object.assign(event, { status: 'failed', retry_count: (event.retry_count || 0) + 1, last_error: JSON.stringify((ack.errors || {})[event.event_id] || 'rejected'), next_retry_at: new Date(Date.now() + 30000).toISOString(), updated_at: now, lease_id: null, lease_until: null });
      });
      store.state.last_checkpoint = ack.server_checkpoint || store.state.last_checkpoint;
      store.state.last_sync_at = now;
      return store;
    });
    return;
  }
  await ensureSchema();
  const { sql } = db();
  await withTransaction(async (tx) => {
    for (const eventId of ack.accepted_event_ids || []) {
      await tx.request()
        .input('event_id', sql.NVarChar, eventId)
        .input('batch_id', sql.NVarChar, ack.batch_id || null)
        .input('lease_id', sql.NVarChar, ack.lease_id || null)
        .query("UPDATE dbo.ArkhonSyncEvents SET status='synced', synced_at=SYSUTCDATETIME(), updated_at=SYSUTCDATETIME(), last_error=NULL, lease_id=NULL, lease_until=NULL WHERE event_id=@event_id AND status='in_flight' AND (@batch_id IS NULL OR batch_id=@batch_id) AND (@lease_id IS NULL OR lease_id=@lease_id)");
    }
    for (const eventId of ack.rejected_event_ids || []) {
      await tx.request()
        .input('event_id', sql.NVarChar, eventId)
        .input('batch_id', sql.NVarChar, ack.batch_id || null)
        .input('error', sql.NVarChar, JSON.stringify((ack.errors || {})[eventId] || 'rejected'))
        .input('lease_id', sql.NVarChar, ack.lease_id || null)
        .query("UPDATE dbo.ArkhonSyncEvents SET status='failed', retry_count=retry_count+1, last_error=@error, next_retry_at=DATEADD(second,30,SYSUTCDATETIME()), lease_id=NULL, lease_until=NULL, updated_at=SYSUTCDATETIME() WHERE event_id=@event_id AND (@batch_id IS NULL OR batch_id=@batch_id) AND (@lease_id IS NULL OR lease_id=@lease_id)");
    }
    if (ack.batch_id) await tx.request().input('batch_id', sql.NVarChar, ack.batch_id).query("UPDATE dbo.ArkhonSyncBatches SET status='acked', updated_at=SYSUTCDATETIME() WHERE batch_id=@batch_id");
    if (ack.server_checkpoint) await setState('last_checkpoint', ack.server_checkpoint, tx);
    await setState('last_sync_at', new Date().toISOString(), tx);
  });
}

async function setState(key, value, connection = null) {
  if (!value) return;
  const { getPool, sql } = db();
  const conn = connection || await getPool();
  await conn.request().input('key', sql.NVarChar, key).input('value', sql.NVarChar(sql.MAX), value).query('MERGE dbo.ArkhonSyncState AS t USING (SELECT @key AS [key], @value AS [value]) AS s ON t.[key]=s.[key] WHEN MATCHED THEN UPDATE SET [value]=s.[value], updated_at=SYSUTCDATETIME() WHEN NOT MATCHED THEN INSERT ([key],[value]) VALUES (s.[key],s.[value]);');
}

async function markFailed(eventIds, error, options = {}) {
  const maxRetries = options.maxRetries || DEFAULT_MAX_RETRIES;
  const baseSeconds = options.baseSeconds || 10;
  if (driver() === 'file') {
    const ids = new Set(eventIds);
    await updateFile((store) => {
      const now = new Date().toISOString();
      store.events.forEach((event) => {
        if (!ids.has(event.event_id)) return;
        const retries = (event.retry_count || 0) + 1;
        event.retry_count = retries;
        event.status = retries >= maxRetries ? 'dead_letter' : 'failed';
        event.last_error = String(error || 'sync_failed').slice(0, 500);
        event.next_retry_at = new Date(Date.now() + Math.min(baseSeconds * (2 ** Math.max(0, retries - 1)), 3600) * 1000).toISOString();
        event.updated_at = now;
      });
      return store;
    });
    return;
  }
  await ensureSchema();
  const { sql } = db();
  await withTransaction(async (tx) => {
    for (const eventId of eventIds) {
      await tx.request()
      .input('event_id', sql.NVarChar, eventId)
      .input('error', sql.NVarChar, String(error || 'sync_failed').slice(0, 500))
      .input('max_retries', sql.Int, maxRetries)
      .input('base_seconds', sql.Int, baseSeconds)
      .query(`UPDATE dbo.ArkhonSyncEvents
SET retry_count=retry_count+1,
    status=CASE WHEN retry_count+1 >= @max_retries THEN 'dead_letter' ELSE 'failed' END,
    last_error=@error,
    next_retry_at=DATEADD(second, IIF(POWER(2, retry_count) * @base_seconds > 3600, 3600, POWER(2, retry_count) * @base_seconds), SYSUTCDATETIME()),
    updated_at=SYSUTCDATETIME()
WHERE event_id=@event_id`);
    }
  });
}

async function markEventSynced(eventId, ack = {}) {
  if (!ack.batch_id) throw new Error('batch_id requerido para ACK de evento');
  await applyAck({ batch_id: ack.batch_id, lease_id: ack.lease_id, accepted_event_ids: [eventId], rejected_event_ids: [], server_checkpoint: ack.server_checkpoint });
}

async function buildBatch(events, context = getSyncContext()) {
  requireLocalIdentity(context);
  return {
    batch_id: crypto.randomUUID(),
    connector_id: context.connectorId,
    device_id: context.deviceId,
    install_id: context.installId,
    protocol_version: 'arkhon-sync-v1',
    client_clock: new Date().toISOString(),
    last_checkpoint: (await getQueueStats()).last_checkpoint,
    events: events.map((event) => {
      assertTenantScope(event, context);
      return event;
    }),
  };
}

async function receiveCommand(command) {
  if (driver() === 'file') {
    let result;
    await updateFile((store) => {
      store.commands = store.commands || [];
      const existing = store.commands.find((item) => item.command_id === command.command_id || item.idempotency_key === command.idempotency_key);
      if (existing) { result = { accepted: false, command: existing }; return store; }
      const item = { command_id: command.command_id, command_type: command.command_type, idempotency_key: command.idempotency_key, payload: sanitizePayload(command.payload || {}), status: 'received', received_at: new Date().toISOString(), error: null };
      store.commands.push(item);
      result = { accepted: true, command: item };
      return store;
    });
    return result;
  }
  await ensureSchema();
  const { sql } = db();
  return withTransaction(async (tx) => {
    const existing = await tx.request().input('id', sql.NVarChar, command.command_id).input('key', sql.NVarChar, command.idempotency_key).query('SELECT * FROM dbo.ArkhonCloudCommands WITH (UPDLOCK, HOLDLOCK) WHERE command_id=@id OR idempotency_key=@key');
    if (existing.recordset.length) return { accepted: false, command: existing.recordset[0] };
    await tx.request().input('id', sql.NVarChar, command.command_id).input('type', sql.NVarChar, command.command_type).input('key', sql.NVarChar, command.idempotency_key).input('payload', sql.NVarChar(sql.MAX), JSON.stringify(sanitizePayload(command.payload || {}))).query("INSERT INTO dbo.ArkhonCloudCommands (command_id, command_type, idempotency_key, payload_json, status) VALUES (@id,@type,@key,@payload,'received')");
    await tx.request().input('event', sql.NVarChar, 'command.received').input('entity_id', sql.NVarChar, command.command_id).input('payload', sql.NVarChar(sql.MAX), JSON.stringify({ command_type: command.command_type })).query('INSERT INTO dbo.ArkhonSyncAudit (event_name, entity_id, payload_json) VALUES (@event,@entity_id,@payload)');
    return { accepted: true, command };
  });
}

async function updateCommand(command, status, fields = {}) {
  if (driver() === 'file') {
    await updateFile((store) => {
      const existing = (store.commands || []).find((item) => item.command_id === command.command_id || item.idempotency_key === command.idempotency_key);
      if (existing) Object.assign(existing, fields, { status, updated_at: new Date().toISOString() });
      return store;
    });
    return;
  }
  await ensureSchema();
  const { sql } = db();
  await withTransaction(async (tx) => {
    await tx.request().input('id', sql.NVarChar, command.command_id).input('status', sql.NVarChar, status).input('result', sql.NVarChar(sql.MAX), fields.result ? JSON.stringify(sanitizePayload(fields.result)) : null).input('error', sql.NVarChar, fields.error || null).query("UPDATE dbo.ArkhonCloudCommands SET status=@status, result_json=@result, error=@error, updated_at=SYSUTCDATETIME(), applying_at=CASE WHEN @status='applying' THEN SYSUTCDATETIME() ELSE applying_at END, applied_at=CASE WHEN @status='applied' THEN SYSUTCDATETIME() ELSE applied_at END WHERE command_id=@id");
    await tx.request().input('event', sql.NVarChar, `command.${status}`).input('entity_id', sql.NVarChar, command.command_id).input('payload', sql.NVarChar(sql.MAX), JSON.stringify({ error: fields.error || null })).query('INSERT INTO dbo.ArkhonSyncAudit (event_name, entity_id, payload_json) VALUES (@event,@entity_id,@payload)');
  });
}

async function markCommandApplying(command) { return updateCommand(command, 'applying'); }
async function markCommandApplied(command, result = {}) { return updateCommand(command, 'applied', { result }); }
async function markCommandError(command, error) { return updateCommand(command, 'failed', { error: String(error || 'command_error').slice(0, 500) }); }

async function getCommandStats() {
  if (driver() === 'file') {
    const commands = readFileStore().commands || [];
    return commands.reduce((acc, command) => { acc[command.status] = (acc[command.status] || 0) + 1; return acc; }, { total: commands.length });
  }
  await ensureSchema();
  const { getPool } = db();
  const pool = await getPool();
  const result = await pool.request().query('SELECT status, COUNT(*) AS count FROM dbo.ArkhonCloudCommands GROUP BY status');
  return result.recordset.reduce((acc, row) => { acc[row.status] = Number(row.count); acc.total += Number(row.count); return acc; }, { total: 0 });
}

module.exports = {
  getStorePath,
  readStore: readFileStore,
  ensureSchema,
  createEvent,
  listEvents,
  getQueueStats,
  dueEvents,
  markSyncing,
  applyAck,
  markFailed,
  markEventSynced,
  buildBatch,
  receiveCommand,
  markCommandApplying,
  markCommandApplied,
  markCommandError,
  getCommandStats,
  recoverExpiredLeases,
};
