feat(sync): add sync queue status observability

cpp/database/DatabaseManager.cpp

@@ -1,5 +1,6 @@
 #include "DatabaseManager.hpp"
 #include "SchemaRegistry.hpp"
+#include "MigrationEngine.hpp"
 #include "../platform/platform.hpp"
 
 namespace margelo::nitro::salvedb {
@@ -10,6 +11,8 @@ void DatabaseManager::open(const std::string& dbName, bool walMode) {
   _db = std::make_shared<SQLiteConnection>(path, walMode);
   // Keyed by schema name, not db file — avoid stale leaks across opens.
   SchemaRegistry::shared().clear();
+  // So sync_queue reads (e.g. getSyncQueueStatus) work even before registerSchema() runs.
+  MigrationEngine::ensureSyncInfra(*_db);
 }
 
 void DatabaseManager::configureCredentials(

cpp/database/HybridSalveDatabase.cpp

@@ -1,6 +1,7 @@
 #include "HybridSalveDatabase.hpp"
 #include "DatabaseManager.hpp"
 #include "MigrationEngine.hpp"
+#include "../sync/SyncQueueReader.hpp"
 #include <stdexcept>
 
 namespace margelo::nitro::salvedb {
@@ -69,6 +70,14 @@ std::shared_ptr<Promise<NativeSyncResult>> HybridSalveDatabase::triggerSync(cons
   });
 }
 
+SyncQueueStatus HybridSalveDatabase::getSyncQueueStatus(const std::string& schemaName) {
+  auto& mgr = DatabaseManager::shared();
+  if (!mgr.isOpen())
+    throw std::runtime_error("Database.getSyncQueueStatus: call Database.configure() before querying sync status");
+  SyncQueueReader reader(mgr.connection());
+  return reader.getStatus(schemaName);
+}
+
 double HybridSalveDatabase::subscribeToChanges(const std::function<void(const std::vector<std::string>&)>& callback) {
   int id = DatabaseManager::shared().connection()->subscribe(
     [callback](std::vector<std::string> tables) { callback(tables); }

cpp/database/HybridSalveDatabase.hpp

@@ -18,6 +18,7 @@ class HybridSalveDatabase: public HybridSalveDatabaseSpec {
   void commit() override;
   void rollback() override;
   std::shared_ptr<Promise<NativeSyncResult>> triggerSync(const std::string& schemaName) override;
+  SyncQueueStatus getSyncQueueStatus(const std::string& schemaName) override;
   double subscribeToChanges(const std::function<void(const std::vector<std::string>&)>& callback) override;
   void unsubscribeFromChanges(double id) override;
   double debugPreparedStatementCount() override;

cpp/database/MigrationEngine.cpp

@@ -92,6 +92,8 @@ static constexpr auto kSyncQueueTable = R"sql(
   );
 )sql";
 
+// Backs SyncQueueReader::getStatus's per-entity pending count and readPage's
+// oldest-first per-entity scan, without a full table scan.
 static constexpr auto kSyncQueueEntityIndex = R"sql(
   CREATE INDEX IF NOT EXISTS idx_sync_queue_entity_id ON sync_queue (entity, id);
 )sql";
@@ -299,16 +301,20 @@ void MigrationEngine::dropSyncTriggers(const SchemaDef& schema) {
 
 // ── Public: registerSchema ────────────────────────────────────────────────────
 
+void MigrationEngine::ensureSyncInfra(SQLiteConnection& db) {
+  db.exec(kSyncQueueTable);
+  db.exec(kSyncQueueEntityIndex);
+  db.exec(kSyncApplyLockTable);
+  db.exec(kSyncCursorTable);
+  db.exec(kSyncDefinitionTable);
+}
+
 void MigrationEngine::registerSchema(const SchemaDef& schema) {
   TransactionGuard txn(*_db);
 
   // Ensure version tracking / sync queue tables exist
   _db->exec(kVersionTable);
-  _db->exec(kSyncQueueTable);
-  _db->exec(kSyncQueueEntityIndex);
-  _db->exec(kSyncApplyLockTable);
-  _db->exec(kSyncCursorTable);
-  _db->exec(kSyncDefinitionTable);
+  ensureSyncInfra(*_db);
 
   int stored = storedVersion(schema.name);
   bool columnsChanged = false;

cpp/database/MigrationEngine.hpp

@@ -48,6 +48,11 @@ class MigrationEngine {
 
   static SchemaDef parseSchemaJson(const std::string& json);
 
+  // Creates sync_queue/_sync_apply_lock/their index if absent (idempotent).
+  // Called eagerly by DatabaseManager::open() so sync-status reads work even
+  // before the first registerSchema() call.
+  static void ensureSyncInfra(SQLiteConnection& db);
+
 private:
   std::shared_ptr<SQLiteConnection> _db;
 

cpp/sync/SyncQueueReader.cpp

@@ -43,6 +43,21 @@ json::Array SyncQueueReader::readOperations(int limit) {
   return operations;
 }
 
+SyncQueueStatus SyncQueueReader::getStatus(const std::string& entity) {
+  auto result = _conn->execute(
+    "SELECT COUNT(*), MIN(updated_at) FROM sync_queue WHERE entity = ?",
+    { entity }
+  );
+
+  double pendingCount = std::get<double>(result.rows[0][0]);
+  std::optional<double> oldestPendingUpdatedAt;
+  if (pendingCount > 0 && std::holds_alternative<double>(result.rows[0][1])) {
+    oldestPendingUpdatedAt = std::get<double>(result.rows[0][1]);
+  }
+
+  return SyncQueueStatus(pendingCount, oldestPendingUpdatedAt);
+}
+
 SyncQueuePage SyncQueueReader::readPage(const std::string& entity, int limit) {
   if (limit < 0) {
     throw std::runtime_error("SyncQueueReader: limit must be >= 0, got " + std::to_string(limit));

cpp/sync/SyncQueueReader.hpp

@@ -2,9 +2,11 @@
 
 #include "../database/SQLiteConnection.hpp"
 #include "../database/json_parser.hpp"
+#include "SyncQueueStatus.hpp"
 #include <cstdint>
 #include <memory>
 #include <optional>
+#include <string>
 
 namespace margelo::nitro::salvedb {
 
@@ -21,6 +23,9 @@ class SyncQueueReader {
   // FIFO, up to `limit` rows, serialized to the TS `ISyncOperation` shape.
   json::Array readOperations(int limit);
 
+  // Pending count + oldest `updated_at` for one entity, no sync involved.
+  SyncQueueStatus getStatus(const std::string& entity);
+
   // Same, filtered to a single entity, plus the row id needed to clear only what was sent.
   SyncQueuePage readPage(const std::string& entity, int limit);
 

cpp/tests/database/HybridSalveDatabaseTests.cpp

@@ -229,3 +229,36 @@ TEST_CASE("blob ArrayBuffer params survive the async JSI thread hop", "[thread-s
     REQUIRE(result == "[1,2,3,4," + std::to_string(i) + "]");
   }
 }
+
+TEST_CASE("getSyncQueueStatus reads pending count/oldest updatedAt through the real JSI bridge", "[sync]") {
+  HybridDatabaseHarness harness;
+  harness.run("(() => { globalThis.db = globalThis.NitroModulesProxy.createHybridObject('SalveDatabase'); return true; })()");
+  harness.run(configureExpr(uniqueDbName("sync_status")));
+
+  harness.run(R"(
+    db.registerSchema(JSON.stringify({
+      name: 'customers', version: 1, primaryKey: 'id',
+      columns: { id: { type: 'integer' }, name: { type: 'text' }, updatedAt: { type: 'datetime', nullable: false } },
+      sync: { enabled: true }
+    }))
+  )");
+
+  auto empty = harness.run("db.getSyncQueueStatus('customers')");
+  REQUIRE(empty == R"({"pendingCount":0})");
+
+  harness.run("db.execute('INSERT INTO customers (id, name, updatedAt) VALUES (1, ?, 100)', ['a'])");
+  harness.run("db.execute('INSERT INTO customers (id, name, updatedAt) VALUES (2, ?, 100)', ['b'])");
+
+  auto pending = harness.run("db.getSyncQueueStatus('customers')");
+  REQUIRE(pending.find(R"("pendingCount":2)") != std::string::npos);
+  REQUIRE(pending.find("oldestPendingUpdatedAt") != std::string::npos);
+}
+
+TEST_CASE("getSyncQueueStatus works right after configure(), before any registerSchema() call", "[sync]") {
+  HybridDatabaseHarness harness;
+  harness.run("(() => { globalThis.db = globalThis.NitroModulesProxy.createHybridObject('SalveDatabase'); return true; })()");
+  harness.run(configureExpr(uniqueDbName("sync_status_no_schema")));
+
+  auto status = harness.run("db.getSyncQueueStatus('customers')");
+  REQUIRE(status == R"({"pendingCount":0})");
+}

cpp/tests/sync/SyncQueueReaderTests.cpp

@@ -97,6 +97,69 @@ TEST_CASE("readOperations with limit 0 returns an empty array", "[sync][SyncQueu
   REQUIRE(ops.empty());
 }
 
+TEST_CASE("getStatus reports zero pending and no oldest timestamp for an empty queue", "[sync][SyncQueueReader]") {
+  auto conn = std::make_shared<SQLiteConnection>(uniqueDbPath("status_empty"));
+  MigrationEngine engine(conn);
+  registerSyncEnabledCustomers(engine);
+
+  SyncQueueReader reader(conn);
+  auto status = reader.getStatus("customers");
+
+  REQUIRE(status.pendingCount == 0);
+  REQUIRE_FALSE(status.oldestPendingUpdatedAt.has_value());
+}
+
+TEST_CASE("getStatus counts pending rows and reports the oldest updated_at", "[sync][SyncQueueReader]") {
+  auto conn = std::make_shared<SQLiteConnection>(uniqueDbPath("status_pending"));
+  MigrationEngine engine(conn);
+  registerSyncEnabledCustomers(engine);
+
+  conn->execute("INSERT INTO customers (id, name, updatedAt) VALUES (1, 'a', 100)", {});
+  conn->execute("INSERT INTO customers (id, name, updatedAt) VALUES (2, 'b', 100)", {});
+
+  SyncQueueReader reader(conn);
+  auto status = reader.getStatus("customers");
+
+  REQUIRE(status.pendingCount == 2);
+  REQUIRE(status.oldestPendingUpdatedAt.has_value());
+
+  auto rows = conn->execute("SELECT MIN(updated_at) FROM sync_queue WHERE entity = 'customers'", {});
+  REQUIRE(*status.oldestPendingUpdatedAt == std::get<double>(rows.rows[0][0]));
+}
+
+TEST_CASE("getStatus only counts rows for the requested entity", "[sync][SyncQueueReader]") {
+  auto conn = std::make_shared<SQLiteConnection>(uniqueDbPath("status_scoped"));
+  MigrationEngine engine(conn);
+  registerSyncEnabledCustomers(engine);
+  engine.registerSchema(MigrationEngine::parseSchemaJson(R"({
+    "name": "orders", "version": 1, "primaryKey": "id",
+    "columns": { "id": { "type": "integer" }, "updatedAt": { "type": "datetime", "nullable": false } },
+    "sync": { "enabled": true }
+  })"));
+
+  conn->execute("INSERT INTO customers (id, name, updatedAt) VALUES (1, 'a', 100)", {});
+  conn->execute("INSERT INTO orders (id, updatedAt) VALUES (1, 100)", {});
+  conn->execute("INSERT INTO orders (id, updatedAt) VALUES (2, 100)", {});
+
+  SyncQueueReader reader(conn);
+  REQUIRE(reader.getStatus("customers").pendingCount == 1);
+  REQUIRE(reader.getStatus("orders").pendingCount == 2);
+}
+
+TEST_CASE("getStatus does not mutate sync_queue", "[sync][SyncQueueReader]") {
+  auto conn = std::make_shared<SQLiteConnection>(uniqueDbPath("status_readonly"));
+  MigrationEngine engine(conn);
+  registerSyncEnabledCustomers(engine);
+
+  conn->execute("INSERT INTO customers (id, name, updatedAt) VALUES (1, 'a', 100)", {});
+
+  SyncQueueReader reader(conn);
+  reader.getStatus("customers");
+
+  auto rows = conn->execute("SELECT COUNT(*) FROM sync_queue", {});
+  REQUIRE(std::get<double>(rows.rows[0][0]) == 1.0);
+}
+
 TEST_CASE("readPage filters by entity", "[sync][SyncQueueReader]") {
   auto conn = std::make_shared<SQLiteConnection>(uniqueDbPath("reader_page_entity"));
   MigrationEngine engine(conn);

docs/mvp-scope.md

@@ -32,6 +32,7 @@ SQLite offline-first pra React Native. Sync engine roda 100% na camada nativa (S
 | **Raw SQL** | `db.execute(sql, params)` como escape hatch, sem type-safety | builder não cobre 100% dos casos; como trigger é a nível de tabela, raw SQL ainda fica rastreado pela sync_queue |
 | **Tipo de `datetime`** | mapeado pra `number` (epoch millis) no TS, não `Date` | mesma convenção já usada em `SyncOperation.updatedAt`, evita ambiguidade de timezone |
 | **Trigger bypass no apply do sync** | tabela `_sync_apply_lock` (1 linha), trigger usa `WHEN NOT EXISTS (SELECT 1 FROM _sync_apply_lock)`. Engine faz `INSERT`/apply/`DELETE` numa única transação. SQL completo em [`query-layer.md`](./query-layer.md#trigger-bypass-durante-apply-do-sync) | sem isso, dado baixado do servidor reentra na sync_queue como se fosse mudança local — loop |
+| **Observabilidade do sync** | `Database.getSyncQueueStatus(schema)` / hook `useSyncStatus(schema)` — lê `pendingCount` + `oldestPendingUpdatedAt` de `sync_queue` por entidade, sem depender do orchestrator (`triggerSync`). Índice em `sync_queue(entity)` evita full scan. Reativo via a mesma assinatura nativa compartilhada do `useQuery` (`QueryCache.subscribeToTables`) | app offline-first precisa mostrar "N pendentes / desde quando" na UI antes mesmo do orchestrator (TASK-012) estar pronto — o dado já existe em `sync_queue`, só faltava expor |
 
 ## Fora do MVP (tipado como futuro, não implementado)
 

src/database/Database.class.ts

@@ -70,6 +70,11 @@ export class Database {
     return this.queryDb.execute(sql, params);
   }
 
+  /** Reads `sync_queue` for `schema` without running a sync — how many writes are pending, and since when. */
+  static getSyncQueueStatus = <TSchema extends AnySchema>(schema: TSchema) => {
+    return this.queryDb.getSyncQueueStatus(schema);
+  }
+
   /**
    * Subscribes to table-level write notifications (insert/update/delete —
    * from any source: query builder, raw SQL, migrations, or background sync).