From b590477d86935cfc119b81a7c4f0d30ed3629dc2 Mon Sep 17 00:00:00 2001 From: eumaninho54 Date: Mon, 13 Jul 2026 18:21:13 -0300 Subject: [PATCH 1/3] feat(sync): add sync queue status observability Database.getSyncQueueStatus(schema) / useSyncStatus(schema) hook read sync_queue's pending count and oldest pending timestamp per entity, without needing the sync orchestrator (still a stub) to be finished. Backed by a new indexed SyncQueueReader::getStatus, with sync_queue's setup tables made eager (DatabaseManager::open()) so status reads work even before the first registerSchema() call. Co-Authored-By: Claude Sonnet 5 --- cpp/database/DatabaseManager.cpp | 3 + cpp/database/HybridSalveDatabase.cpp | 9 ++ cpp/database/HybridSalveDatabase.hpp | 1 + cpp/database/MigrationEngine.cpp | 14 +- cpp/database/MigrationEngine.hpp | 5 + cpp/sync/SyncQueueReader.cpp | 15 ++ cpp/sync/SyncQueueReader.hpp | 5 + .../database/HybridSalveDatabaseTests.cpp | 33 +++++ cpp/tests/sync/SyncQueueReaderTests.cpp | 63 +++++++++ docs/mvp-scope.md | 1 + .../shared/c++/HybridSalveDatabaseSpec.cpp | 1 + .../shared/c++/HybridSalveDatabaseSpec.hpp | 4 + .../generated/shared/c++/SyncQueueStatus.hpp | 87 ++++++++++++ src/database/Database.class.ts | 5 + src/database/classes/QueryDb/QueryDb.class.ts | 7 + src/hooks/index.ts | 1 + .../__tests__/useSyncStatus.test.tsx | 133 ++++++++++++++++++ src/hooks/useSyncStatus/index.ts | 64 +++++++++ .../types/IUseSyncStatusResult.ts | 8 ++ src/hooks/useSyncStatus/types/index.ts | 1 + src/specs/SalveDatabase.nitro.ts | 8 ++ src/types/sync/SyncQueueStatus.ts | 8 ++ src/types/sync/index.ts | 1 + 23 files changed, 475 insertions(+), 2 deletions(-) create mode 100644 nitrogen/generated/shared/c++/SyncQueueStatus.hpp create mode 100644 src/hooks/useSyncStatus/__tests__/useSyncStatus.test.tsx create mode 100644 src/hooks/useSyncStatus/index.ts create mode 100644 src/hooks/useSyncStatus/types/IUseSyncStatusResult.ts create mode 100644 src/hooks/useSyncStatus/types/index.ts create mode 100644 src/types/sync/SyncQueueStatus.ts diff --git a/cpp/database/DatabaseManager.cpp b/cpp/database/DatabaseManager.cpp index 1521077..6419ea2 100644 --- a/cpp/database/DatabaseManager.cpp +++ b/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 { @@ -11,6 +12,8 @@ void DatabaseManager::open(const std::string& dbName, bool walMode) { // Boolean-column registrations are keyed by table name, not by db file — a stale // entry from a previously-open database would otherwise silently leak into this one. SchemaRegistry::shared().clear(); + // So sync_queue reads (e.g. getSyncQueueStatus) work even before registerSchema() runs. + MigrationEngine::ensureSyncInfra(*_db); } void DatabaseManager::configureCredentials( diff --git a/cpp/database/HybridSalveDatabase.cpp b/cpp/database/HybridSalveDatabase.cpp index 9b2b727..a1544fd 100644 --- a/cpp/database/HybridSalveDatabase.cpp +++ b/cpp/database/HybridSalveDatabase.cpp @@ -1,6 +1,7 @@ #include "HybridSalveDatabase.hpp" #include "DatabaseManager.hpp" #include "MigrationEngine.hpp" +#include "../sync/SyncQueueReader.hpp" #include namespace margelo::nitro::salvedb { @@ -69,6 +70,14 @@ std::shared_ptr> 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&)>& callback) { int id = DatabaseManager::shared().connection()->subscribe( [callback](std::vector tables) { callback(tables); } diff --git a/cpp/database/HybridSalveDatabase.hpp b/cpp/database/HybridSalveDatabase.hpp index f156775..ef1a7aa 100644 --- a/cpp/database/HybridSalveDatabase.hpp +++ b/cpp/database/HybridSalveDatabase.hpp @@ -18,6 +18,7 @@ class HybridSalveDatabase: public HybridSalveDatabaseSpec { void commit() override; void rollback() override; std::shared_ptr> triggerSync(const std::string& schemaName) override; + SyncQueueStatus getSyncQueueStatus(const std::string& schemaName) override; double subscribeToChanges(const std::function&)>& callback) override; void unsubscribeFromChanges(double id) override; double debugPreparedStatementCount() override; diff --git a/cpp/database/MigrationEngine.cpp b/cpp/database/MigrationEngine.cpp index ba31b29..757f55c 100644 --- a/cpp/database/MigrationEngine.cpp +++ b/cpp/database/MigrationEngine.cpp @@ -97,6 +97,11 @@ static constexpr auto kSyncApplyLockTable = R"sql( ); )sql"; +// Backs SyncQueueReader::getStatus's per-entity pending count without a full table scan. +static constexpr auto kSyncQueueEntityIndex = R"sql( + CREATE INDEX IF NOT EXISTS idx_sync_queue_entity ON sync_queue(entity); +)sql"; + int MigrationEngine::storedVersion(const std::string& schemaName) { auto result = _db->execute( "SELECT version FROM _salve_schema_versions WHERE name = ?", @@ -280,13 +285,18 @@ void MigrationEngine::dropSyncTriggers(const SchemaDef& schema) { // ── Public: registerSchema ──────────────────────────────────────────────────── +void MigrationEngine::ensureSyncInfra(SQLiteConnection& db) { + db.exec(kSyncQueueTable); + db.exec(kSyncApplyLockTable); + db.exec(kSyncQueueEntityIndex); +} + void MigrationEngine::registerSchema(const SchemaDef& schema) { TransactionGuard txn(*_db); // Ensure version tracking / sync queue tables exist _db->exec(kVersionTable); - _db->exec(kSyncQueueTable); - _db->exec(kSyncApplyLockTable); + ensureSyncInfra(*_db); int stored = storedVersion(schema.name); bool columnsChanged = false; diff --git a/cpp/database/MigrationEngine.hpp b/cpp/database/MigrationEngine.hpp index 671ed6a..3e1e2b4 100644 --- a/cpp/database/MigrationEngine.hpp +++ b/cpp/database/MigrationEngine.hpp @@ -46,6 +46,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 _db; diff --git a/cpp/sync/SyncQueueReader.cpp b/cpp/sync/SyncQueueReader.cpp index 1609a72..57f2490 100644 --- a/cpp/sync/SyncQueueReader.cpp +++ b/cpp/sync/SyncQueueReader.cpp @@ -33,4 +33,19 @@ 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(result.rows[0][0]); + std::optional oldestPendingUpdatedAt; + if (pendingCount > 0 && std::holds_alternative(result.rows[0][1])) { + oldestPendingUpdatedAt = std::get(result.rows[0][1]); + } + + return SyncQueueStatus(pendingCount, oldestPendingUpdatedAt); +} + } // namespace margelo::nitro::salvedb diff --git a/cpp/sync/SyncQueueReader.hpp b/cpp/sync/SyncQueueReader.hpp index 8a386f5..54dd16f 100644 --- a/cpp/sync/SyncQueueReader.hpp +++ b/cpp/sync/SyncQueueReader.hpp @@ -2,7 +2,9 @@ #include "../database/SQLiteConnection.hpp" #include "../database/json_parser.hpp" +#include "SyncQueueStatus.hpp" #include +#include namespace margelo::nitro::salvedb { @@ -14,6 +16,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); + private: std::shared_ptr _conn; }; diff --git a/cpp/tests/database/HybridSalveDatabaseTests.cpp b/cpp/tests/database/HybridSalveDatabaseTests.cpp index aa08d3b..d74fdbc 100644 --- a/cpp/tests/database/HybridSalveDatabaseTests.cpp +++ b/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' } }, + sync: { enabled: true } + })) + )"); + + auto empty = harness.run("db.getSyncQueueStatus('customers')"); + REQUIRE(empty == R"({"pendingCount":0})"); + + harness.run("db.execute('INSERT INTO customers (id, name) VALUES (1, ?)', ['a'])"); + harness.run("db.execute('INSERT INTO customers (id, name) VALUES (2, ?)', ['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})"); +} diff --git a/cpp/tests/sync/SyncQueueReaderTests.cpp b/cpp/tests/sync/SyncQueueReaderTests.cpp index 8dc0791..0c8d81e 100644 --- a/cpp/tests/sync/SyncQueueReaderTests.cpp +++ b/cpp/tests/sync/SyncQueueReaderTests.cpp @@ -96,3 +96,66 @@ TEST_CASE("readOperations with limit 0 returns an empty array", "[sync][SyncQueu auto ops = reader.readOperations(0); REQUIRE(ops.empty()); } + +TEST_CASE("getStatus reports zero pending and no oldest timestamp for an empty queue", "[sync][SyncQueueReader]") { + auto conn = std::make_shared(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(uniqueDbPath("status_pending")); + MigrationEngine engine(conn); + registerSyncEnabledCustomers(engine); + + conn->execute("INSERT INTO customers (id, name) VALUES (1, 'a')", {}); + conn->execute("INSERT INTO customers (id, name) VALUES (2, 'b')", {}); + + 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(rows.rows[0][0])); +} + +TEST_CASE("getStatus only counts rows for the requested entity", "[sync][SyncQueueReader]") { + auto conn = std::make_shared(uniqueDbPath("status_scoped")); + MigrationEngine engine(conn); + registerSyncEnabledCustomers(engine); + engine.registerSchema(MigrationEngine::parseSchemaJson(R"({ + "name": "orders", "version": 1, "primaryKey": "id", + "columns": { "id": { "type": "integer" } }, + "sync": { "enabled": true } + })")); + + conn->execute("INSERT INTO customers (id, name) VALUES (1, 'a')", {}); + conn->execute("INSERT INTO orders (id) VALUES (1)", {}); + conn->execute("INSERT INTO orders (id) VALUES (2)", {}); + + 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(uniqueDbPath("status_readonly")); + MigrationEngine engine(conn); + registerSyncEnabledCustomers(engine); + + conn->execute("INSERT INTO customers (id, name) VALUES (1, 'a')", {}); + + SyncQueueReader reader(conn); + reader.getStatus("customers"); + + auto rows = conn->execute("SELECT COUNT(*) FROM sync_queue", {}); + REQUIRE(std::get(rows.rows[0][0]) == 1.0); +} diff --git a/docs/mvp-scope.md b/docs/mvp-scope.md index 251f33d..93247d9 100644 --- a/docs/mvp-scope.md +++ b/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) diff --git a/nitrogen/generated/shared/c++/HybridSalveDatabaseSpec.cpp b/nitrogen/generated/shared/c++/HybridSalveDatabaseSpec.cpp index 99f4f48..82cebea 100644 --- a/nitrogen/generated/shared/c++/HybridSalveDatabaseSpec.cpp +++ b/nitrogen/generated/shared/c++/HybridSalveDatabaseSpec.cpp @@ -21,6 +21,7 @@ namespace margelo::nitro::salvedb { prototype.registerHybridMethod("commit", &HybridSalveDatabaseSpec::commit); prototype.registerHybridMethod("rollback", &HybridSalveDatabaseSpec::rollback); prototype.registerHybridMethod("triggerSync", &HybridSalveDatabaseSpec::triggerSync); + prototype.registerHybridMethod("getSyncQueueStatus", &HybridSalveDatabaseSpec::getSyncQueueStatus); prototype.registerHybridMethod("subscribeToChanges", &HybridSalveDatabaseSpec::subscribeToChanges); prototype.registerHybridMethod("unsubscribeFromChanges", &HybridSalveDatabaseSpec::unsubscribeFromChanges); prototype.registerHybridMethod("debugPreparedStatementCount", &HybridSalveDatabaseSpec::debugPreparedStatementCount); diff --git a/nitrogen/generated/shared/c++/HybridSalveDatabaseSpec.hpp b/nitrogen/generated/shared/c++/HybridSalveDatabaseSpec.hpp index 1e987e0..e9f1db3 100644 --- a/nitrogen/generated/shared/c++/HybridSalveDatabaseSpec.hpp +++ b/nitrogen/generated/shared/c++/HybridSalveDatabaseSpec.hpp @@ -19,6 +19,8 @@ namespace margelo::nitro::salvedb { struct ConfigureParams; } namespace margelo::nitro::salvedb { struct QueryResult; } // Forward declaration of `NativeSyncResult` to properly resolve imports. namespace margelo::nitro::salvedb { struct NativeSyncResult; } +// Forward declaration of `SyncQueueStatus` to properly resolve imports. +namespace margelo::nitro::salvedb { struct SyncQueueStatus; } #include "ConfigureParams.hpp" #include @@ -29,6 +31,7 @@ namespace margelo::nitro::salvedb { struct NativeSyncResult; } #include #include #include "NativeSyncResult.hpp" +#include "SyncQueueStatus.hpp" #include namespace margelo::nitro::salvedb { @@ -69,6 +72,7 @@ namespace margelo::nitro::salvedb { virtual void commit() = 0; virtual void rollback() = 0; virtual std::shared_ptr> triggerSync(const std::string& schemaName) = 0; + virtual SyncQueueStatus getSyncQueueStatus(const std::string& schemaName) = 0; virtual double subscribeToChanges(const std::function& /* tables */)>& callback) = 0; virtual void unsubscribeFromChanges(double id) = 0; virtual double debugPreparedStatementCount() = 0; diff --git a/nitrogen/generated/shared/c++/SyncQueueStatus.hpp b/nitrogen/generated/shared/c++/SyncQueueStatus.hpp new file mode 100644 index 0000000..c2f7dfe --- /dev/null +++ b/nitrogen/generated/shared/c++/SyncQueueStatus.hpp @@ -0,0 +1,87 @@ +/// +/// SyncQueueStatus.hpp +/// This file was generated by nitrogen. DO NOT MODIFY THIS FILE. +/// https://github.com/mrousavy/nitro +/// Copyright © Marc Rousavy @ Margelo +/// + +#pragma once + +#if __has_include() +#include +#else +#error NitroModules cannot be found! Are you sure you installed NitroModules properly? +#endif +#if __has_include() +#include +#else +#error NitroModules cannot be found! Are you sure you installed NitroModules properly? +#endif +#if __has_include() +#include +#else +#error NitroModules cannot be found! Are you sure you installed NitroModules properly? +#endif +#if __has_include() +#include +#else +#error NitroModules cannot be found! Are you sure you installed NitroModules properly? +#endif + + + +#include + +namespace margelo::nitro::salvedb { + + /** + * A struct which can be represented as a JavaScript object (SyncQueueStatus). + */ + struct SyncQueueStatus final { + public: + double pendingCount SWIFT_PRIVATE; + std::optional oldestPendingUpdatedAt SWIFT_PRIVATE; + + public: + SyncQueueStatus() = default; + explicit SyncQueueStatus(double pendingCount, std::optional oldestPendingUpdatedAt): pendingCount(pendingCount), oldestPendingUpdatedAt(oldestPendingUpdatedAt) {} + + public: + friend bool operator==(const SyncQueueStatus& lhs, const SyncQueueStatus& rhs) = default; + }; + +} // namespace margelo::nitro::salvedb + +namespace margelo::nitro { + + // C++ SyncQueueStatus <> JS SyncQueueStatus (object) + template <> + struct JSIConverter final { + static inline margelo::nitro::salvedb::SyncQueueStatus fromJSI(jsi::Runtime& runtime, const jsi::Value& arg) { + jsi::Object obj = arg.asObject(runtime); + return margelo::nitro::salvedb::SyncQueueStatus( + JSIConverter::fromJSI(runtime, obj.getProperty(runtime, PropNameIDCache::get(runtime, "pendingCount"))), + JSIConverter>::fromJSI(runtime, obj.getProperty(runtime, PropNameIDCache::get(runtime, "oldestPendingUpdatedAt"))) + ); + } + static inline jsi::Value toJSI(jsi::Runtime& runtime, const margelo::nitro::salvedb::SyncQueueStatus& arg) { + jsi::Object obj(runtime); + obj.setProperty(runtime, PropNameIDCache::get(runtime, "pendingCount"), JSIConverter::toJSI(runtime, arg.pendingCount)); + obj.setProperty(runtime, PropNameIDCache::get(runtime, "oldestPendingUpdatedAt"), JSIConverter>::toJSI(runtime, arg.oldestPendingUpdatedAt)); + return obj; + } + static inline bool canConvert(jsi::Runtime& runtime, const jsi::Value& value) { + if (!value.isObject()) { + return false; + } + jsi::Object obj = value.getObject(runtime); + if (!nitro::isPlainObject(runtime, obj)) { + return false; + } + if (!JSIConverter::canConvert(runtime, obj.getProperty(runtime, PropNameIDCache::get(runtime, "pendingCount")))) return false; + if (!JSIConverter>::canConvert(runtime, obj.getProperty(runtime, PropNameIDCache::get(runtime, "oldestPendingUpdatedAt")))) return false; + return true; + } + }; + +} // namespace margelo::nitro diff --git a/src/database/Database.class.ts b/src/database/Database.class.ts index ca2ddc9..c3afe50 100644 --- a/src/database/Database.class.ts +++ b/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 = (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). diff --git a/src/database/classes/QueryDb/QueryDb.class.ts b/src/database/classes/QueryDb/QueryDb.class.ts index 9ca8d24..46c3233 100644 --- a/src/database/classes/QueryDb/QueryDb.class.ts +++ b/src/database/classes/QueryDb/QueryDb.class.ts @@ -2,6 +2,7 @@ import type { SalveDatabase } from '../../../specs/SalveDatabase.nitro'; import type { SqlValue } from '../../../specs/types'; import type { IQueryClient } from './types'; import type { AnySchema } from '../../../types'; +import type { SyncQueueStatus } from '../../../types/sync/SyncQueueStatus'; import { SelectQueryBuilder, InsertQueryBuilder, UpdateQueryBuilder, DeleteQueryBuilder, CountQueryBuilder } from './classes'; import { ConfigureDb } from '../ConfigureDb'; @@ -58,6 +59,12 @@ export class QueryDb { }) } + /** Reads `sync_queue` for `schema` without running a sync — how many writes are pending, and since when. */ + getSyncQueueStatus(schema: TSchema): SyncQueueStatus { + this._assertConfigured('getSyncQueueStatus'); + return this._bridge.getSyncQueueStatus(schema.name); + } + subscribeToChanges(callback: (tables: string[]) => void): number { this._assertConfigured('subscribeToChanges'); return this._bridge.subscribeToChanges(callback); diff --git a/src/hooks/index.ts b/src/hooks/index.ts index f24fe29..c1c3674 100644 --- a/src/hooks/index.ts +++ b/src/hooks/index.ts @@ -1,3 +1,4 @@ export * from './useDatabaseReady'; export * from './useQuery'; export * from './useInfiniteQuery'; +export * from './useSyncStatus'; diff --git a/src/hooks/useSyncStatus/__tests__/useSyncStatus.test.tsx b/src/hooks/useSyncStatus/__tests__/useSyncStatus.test.tsx new file mode 100644 index 0000000..8efca0e --- /dev/null +++ b/src/hooks/useSyncStatus/__tests__/useSyncStatus.test.tsx @@ -0,0 +1,133 @@ +import type { AnySchema } from '../../../types'; +import type { SyncQueueStatus } from '../../../types/sync/SyncQueueStatus'; +import { SalveDbContext } from '../../../provider/SalveDbContext'; +import type { IDatabaseReadyState } from '../../../provider/types'; +import React from 'react'; +import { act, renderHook } from '@testing-library/react-native'; + +const mockGetSyncQueueStatus = jest.fn(); + +jest.mock('../../../database', () => ({ + Database: { + getSyncQueueStatus: (schema: AnySchema) => mockGetSyncQueueStatus(schema), + }, +})); + +let capturedTables: string[] | null = null; +const mockUnsubscribeFromTables = jest.fn(); +const mockSubscribeToTables = jest.fn((tables: string[], _listener: () => void) => { + capturedTables = tables; + return mockUnsubscribeFromTables; +}); +let capturedListener: (() => void) | null = null; + +jest.mock('../../../cache', () => ({ + queryCache: { + subscribeToTables: (tables: string[], listener: () => void) => { + capturedListener = listener; + return mockSubscribeToTables(tables, listener); + }, + }, +})); + +const { useSyncStatus } = require('../index') as typeof import('../index'); + +const schema: AnySchema = { + name: 'customers', + version: 1, + primaryKey: 'id', + columns: { id: { type: 'integer' } }, +}; + +function withDbState(state: IDatabaseReadyState) { + return function Wrapper({ children }: { children: React.ReactNode }) { + return {children}; + }; +} + +const readyState: IDatabaseReadyState = { isReady: true, isLoading: false, error: null }; +const notReadyState: IDatabaseReadyState = { isReady: false, isLoading: true, error: null }; + +beforeEach(() => { + jest.clearAllMocks(); + capturedTables = null; + capturedListener = null; +}); + +describe('useSyncStatus — readiness gating', () => { + test('does not query while the db is not ready', async () => { + const { result } = await renderHook(() => useSyncStatus(schema), { wrapper: withDbState(notReadyState) }); + + expect(mockGetSyncQueueStatus).not.toHaveBeenCalled(); + expect(result.current).toEqual({ status: null, error: null, isLoading: true }); + }); +}); + +describe('useSyncStatus — once ready', () => { + test('reads the status for the given schema', async () => { + const status: SyncQueueStatus = { pendingCount: 3, oldestPendingUpdatedAt: 1700000000000 }; + mockGetSyncQueueStatus.mockReturnValue(status); + + const { result } = await renderHook(() => useSyncStatus(schema), { wrapper: withDbState(readyState) }); + + expect(result.current).toEqual({ status, error: null, isLoading: false }); + expect(mockGetSyncQueueStatus).toHaveBeenCalledWith(schema); + }); + + test('subscribes to the shared queryCache "sync_queue" table subscription', async () => { + mockGetSyncQueueStatus.mockReturnValue({ pendingCount: 0 }); + + await renderHook(() => useSyncStatus(schema), { wrapper: withDbState(readyState) }); + + expect(mockSubscribeToTables).toHaveBeenCalledTimes(1); + expect(capturedTables).toEqual(['sync_queue']); + }); + + test('an inline schema literal recreated every render does not cause a reload loop', async () => { + mockGetSyncQueueStatus.mockReturnValue({ pendingCount: 1 }); + + const { result } = await renderHook( + () => useSyncStatus({ name: 'customers', version: 1, primaryKey: 'id', columns: { id: { type: 'integer' } } } as AnySchema), + { wrapper: withDbState(readyState) }, + ); + + expect(result.current.status).toEqual({ pendingCount: 1 }); + expect(mockGetSyncQueueStatus).toHaveBeenCalledTimes(1); + expect(mockSubscribeToTables).toHaveBeenCalledTimes(1); + }); + + test('a write to sync_queue re-reads the status', async () => { + mockGetSyncQueueStatus.mockReturnValueOnce({ pendingCount: 1 }); + + const { result } = await renderHook(() => useSyncStatus(schema), { wrapper: withDbState(readyState) }); + expect(result.current.status).toEqual({ pendingCount: 1 }); + + mockGetSyncQueueStatus.mockReturnValueOnce({ pendingCount: 2, oldestPendingUpdatedAt: 42 }); + await act(() => { capturedListener?.(); }); + + expect(result.current.status).toEqual({ pendingCount: 2, oldestPendingUpdatedAt: 42 }); + }); + + test('a transient read error preserves the last known status instead of blanking it', async () => { + mockGetSyncQueueStatus.mockReturnValueOnce({ pendingCount: 1 }); + + const { result } = await renderHook(() => useSyncStatus(schema), { wrapper: withDbState(readyState) }); + expect(result.current.status).toEqual({ pendingCount: 1 }); + + const boom = new Error('boom'); + mockGetSyncQueueStatus.mockImplementationOnce(() => { throw boom; }); + await act(() => { capturedListener?.(); }); + + expect(result.current.status).toEqual({ pendingCount: 1 }); + expect(result.current.error).toBe(boom); + }); + + test('unsubscribes from the shared table subscription on unmount', async () => { + mockGetSyncQueueStatus.mockReturnValue({ pendingCount: 0 }); + + const { unmount } = await renderHook(() => useSyncStatus(schema), { wrapper: withDbState(readyState) }); + await act(() => { unmount(); }); + + expect(mockUnsubscribeFromTables).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/hooks/useSyncStatus/index.ts b/src/hooks/useSyncStatus/index.ts new file mode 100644 index 0000000..fb4eb0e --- /dev/null +++ b/src/hooks/useSyncStatus/index.ts @@ -0,0 +1,64 @@ +import type { AnySchema } from '../../types'; +import type { IUseSyncStatusResult } from './types'; +import { useCallback, useEffect, useRef, useState } from 'react'; +import { Database } from '../../database'; +import { queryCache } from '../../cache'; +import { useDatabaseReady } from '../useDatabaseReady'; + +export type { IUseSyncStatusResult } from './types'; + +interface State { + schemaName: string | null; + status: IUseSyncStatusResult['status']; + error: unknown; +} + +/** + * Reads `sync_queue` for `schema` — how many local writes haven't been sent + * to the server yet, and since when — without running a sync. Kept live: + * any write to `schema`'s table (which also writes to `sync_queue` via the + * trigger engine) refreshes it automatically. + */ +export const useSyncStatus = (schema: TSchema): IUseSyncStatusResult => { + const { isReady, isLoading: dbLoading, error: dbError } = useDatabaseReady(); + + // Keyed off `schema.name` (a stable primitive), not `schema` itself — a caller passing an + // inline schema literal would otherwise recreate `reload`'s identity every render, retriggering + // the effect below in a loop. `latest` still tracks the full schema object so `reload` calls + // the bridge with whatever was passed most recently. + const latest = useRef(schema); + latest.current = schema; + + const [state, setState] = useState({ schemaName: null, status: null, error: null }); + + const reload = useCallback(() => { + try { + const status = Database.getSyncQueueStatus(latest.current); + setState({ schemaName: latest.current.name, status, error: null }); + } catch (error) { + // Keep whatever was already on screen — a transient read failure shouldn't blank a known status. + setState((prev) => ({ ...prev, error })); + } + }, [schema.name]); + + const reloadRef = useRef(reload); + reloadRef.current = reload; + + useEffect(() => { + if (!isReady) return; + reload(); + }, [isReady, reload]); + + useEffect(() => { + if (!isReady) return; + return queryCache.subscribeToTables(['sync_queue'], () => reloadRef.current()); + }, [isReady, schema.name]); + + const isCurrent = isReady && state.schemaName === schema.name; + + return { + status: isCurrent ? state.status : null, + error: dbError ?? (isCurrent ? state.error : null), + isLoading: dbLoading || (isReady && !isCurrent), + }; +}; diff --git a/src/hooks/useSyncStatus/types/IUseSyncStatusResult.ts b/src/hooks/useSyncStatus/types/IUseSyncStatusResult.ts new file mode 100644 index 0000000..05c8c1a --- /dev/null +++ b/src/hooks/useSyncStatus/types/IUseSyncStatusResult.ts @@ -0,0 +1,8 @@ +import type { SyncQueueStatus } from "../../../types/sync/SyncQueueStatus"; + +export interface IUseSyncStatusResult { + /** `null` until the first read completes. */ + status: SyncQueueStatus | null; + error: unknown; + isLoading: boolean; +} diff --git a/src/hooks/useSyncStatus/types/index.ts b/src/hooks/useSyncStatus/types/index.ts new file mode 100644 index 0000000..6a9f7c8 --- /dev/null +++ b/src/hooks/useSyncStatus/types/index.ts @@ -0,0 +1 @@ +export type * from './IUseSyncStatusResult'; diff --git a/src/specs/SalveDatabase.nitro.ts b/src/specs/SalveDatabase.nitro.ts index bbfd31b..5f343ba 100644 --- a/src/specs/SalveDatabase.nitro.ts +++ b/src/specs/SalveDatabase.nitro.ts @@ -1,6 +1,7 @@ import type { HybridObject } from "react-native-nitro-modules"; import type { ConfigureParams, SqlValue, QueryResult } from "./types"; import type { NativeSyncResult } from "../types/sync/NativeSyncResult"; +import type { SyncQueueStatus } from "../types/sync/SyncQueueStatus"; export interface SalveDatabase extends HybridObject<{ ios: "c++"; android: "c++" }> { // ── Lifecycle ────────────────────────────────────────────────────────────── @@ -40,6 +41,13 @@ export interface SalveDatabase extends HybridObject<{ ios: "c++"; android: "c++" */ triggerSync(schemaName: string): Promise; + /** + * Reads `sync_queue` for `schemaName` without running a sync — how many + * writes are waiting to go out, and since when. Synchronous: a single + * indexed `COUNT`/`MIN`, no network involved. + */ + getSyncQueueStatus(schemaName: string): SyncQueueStatus; + // ── Change notification ───────────────────────────────────────────────────── /** diff --git a/src/types/sync/SyncQueueStatus.ts b/src/types/sync/SyncQueueStatus.ts new file mode 100644 index 0000000..7ca3f11 --- /dev/null +++ b/src/types/sync/SyncQueueStatus.ts @@ -0,0 +1,8 @@ +/** Snapshot of `sync_queue` state for one schema, read without running a sync. */ +export interface SyncQueueStatus { + /** Rows in `sync_queue` for this schema, awaiting the next sync session. */ + pendingCount: number; + + /** `updatedAt` of the oldest pending row (epoch millis), absent if `pendingCount` is 0. */ + oldestPendingUpdatedAt?: number; +} diff --git a/src/types/sync/index.ts b/src/types/sync/index.ts index f14e2cb..580f1e4 100644 --- a/src/types/sync/index.ts +++ b/src/types/sync/index.ts @@ -13,4 +13,5 @@ export type * from './ISyncDefinition'; export type * from './SyncDirection'; export type * from './ISyncOperation'; export type * from './SyncStrategy'; +export type * from './SyncQueueStatus'; export type * from './ITransport'; From 3faac83dc1a9357b128216d85cb6125cad7409ec Mon Sep 17 00:00:00 2001 From: eumaninho54 Date: Mon, 13 Jul 2026 21:48:23 -0300 Subject: [PATCH 2/3] fix(test): satisfy sync updatedAt constraint in status tests getSyncQueueStatus/getStatus tests inserted rows without updatedAt, which now fails registerSchema's NOT NULL datetime requirement for sync-enabled schemas (brought in by the develop merge). Co-Authored-By: Claude Sonnet 5 --- cpp/tests/database/HybridSalveDatabaseTests.cpp | 6 +++--- cpp/tests/sync/SyncQueueReaderTests.cpp | 14 +++++++------- 2 files changed, 10 insertions(+), 10 deletions(-) diff --git a/cpp/tests/database/HybridSalveDatabaseTests.cpp b/cpp/tests/database/HybridSalveDatabaseTests.cpp index d74fdbc..44ff566 100644 --- a/cpp/tests/database/HybridSalveDatabaseTests.cpp +++ b/cpp/tests/database/HybridSalveDatabaseTests.cpp @@ -238,7 +238,7 @@ TEST_CASE("getSyncQueueStatus reads pending count/oldest updatedAt through the r harness.run(R"( db.registerSchema(JSON.stringify({ name: 'customers', version: 1, primaryKey: 'id', - columns: { id: { type: 'integer' }, name: { type: 'text' } }, + columns: { id: { type: 'integer' }, name: { type: 'text' }, updatedAt: { type: 'datetime', nullable: false } }, sync: { enabled: true } })) )"); @@ -246,8 +246,8 @@ TEST_CASE("getSyncQueueStatus reads pending count/oldest updatedAt through the r auto empty = harness.run("db.getSyncQueueStatus('customers')"); REQUIRE(empty == R"({"pendingCount":0})"); - harness.run("db.execute('INSERT INTO customers (id, name) VALUES (1, ?)', ['a'])"); - harness.run("db.execute('INSERT INTO customers (id, name) VALUES (2, ?)', ['b'])"); + 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); diff --git a/cpp/tests/sync/SyncQueueReaderTests.cpp b/cpp/tests/sync/SyncQueueReaderTests.cpp index 742742d..fc56aad 100644 --- a/cpp/tests/sync/SyncQueueReaderTests.cpp +++ b/cpp/tests/sync/SyncQueueReaderTests.cpp @@ -114,8 +114,8 @@ TEST_CASE("getStatus counts pending rows and reports the oldest updated_at", "[s MigrationEngine engine(conn); registerSyncEnabledCustomers(engine); - conn->execute("INSERT INTO customers (id, name) VALUES (1, 'a')", {}); - conn->execute("INSERT INTO customers (id, name) VALUES (2, 'b')", {}); + 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"); @@ -133,13 +133,13 @@ TEST_CASE("getStatus only counts rows for the requested entity", "[sync][SyncQue registerSyncEnabledCustomers(engine); engine.registerSchema(MigrationEngine::parseSchemaJson(R"({ "name": "orders", "version": 1, "primaryKey": "id", - "columns": { "id": { "type": "integer" } }, + "columns": { "id": { "type": "integer" }, "updatedAt": { "type": "datetime", "nullable": false } }, "sync": { "enabled": true } })")); - conn->execute("INSERT INTO customers (id, name) VALUES (1, 'a')", {}); - conn->execute("INSERT INTO orders (id) VALUES (1)", {}); - conn->execute("INSERT INTO orders (id) VALUES (2)", {}); + 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); @@ -151,7 +151,7 @@ TEST_CASE("getStatus does not mutate sync_queue", "[sync][SyncQueueReader]") { MigrationEngine engine(conn); registerSyncEnabledCustomers(engine); - conn->execute("INSERT INTO customers (id, name) VALUES (1, 'a')", {}); + conn->execute("INSERT INTO customers (id, name, updatedAt) VALUES (1, 'a', 100)", {}); SyncQueueReader reader(conn); reader.getStatus("customers"); From bb0889ccaa457a7897184672331ff8f7e605f53c Mon Sep 17 00:00:00 2001 From: eumaninho54 Date: Mon, 13 Jul 2026 21:50:28 -0300 Subject: [PATCH 3/3] refactor(hooks): move useSyncStatus's State type to types/ Follows this codebase's convention (e.g. useInfiniteQuery/types/IState.ts) of keeping all typings under a hook's types/ folder, not inline. Co-Authored-By: Claude Sonnet 5 --- src/hooks/useSyncStatus/index.ts | 15 ++------------- src/hooks/useSyncStatus/types/IState.ts | 7 +++++++ src/hooks/useSyncStatus/types/index.ts | 1 + 3 files changed, 10 insertions(+), 13 deletions(-) create mode 100644 src/hooks/useSyncStatus/types/IState.ts diff --git a/src/hooks/useSyncStatus/index.ts b/src/hooks/useSyncStatus/index.ts index fb4eb0e..b06c56e 100644 --- a/src/hooks/useSyncStatus/index.ts +++ b/src/hooks/useSyncStatus/index.ts @@ -1,5 +1,5 @@ import type { AnySchema } from '../../types'; -import type { IUseSyncStatusResult } from './types'; +import type { IState, IUseSyncStatusResult } from './types'; import { useCallback, useEffect, useRef, useState } from 'react'; import { Database } from '../../database'; import { queryCache } from '../../cache'; @@ -7,12 +7,6 @@ import { useDatabaseReady } from '../useDatabaseReady'; export type { IUseSyncStatusResult } from './types'; -interface State { - schemaName: string | null; - status: IUseSyncStatusResult['status']; - error: unknown; -} - /** * Reads `sync_queue` for `schema` — how many local writes haven't been sent * to the server yet, and since when — without running a sync. Kept live: @@ -22,21 +16,16 @@ interface State { export const useSyncStatus = (schema: TSchema): IUseSyncStatusResult => { const { isReady, isLoading: dbLoading, error: dbError } = useDatabaseReady(); - // Keyed off `schema.name` (a stable primitive), not `schema` itself — a caller passing an - // inline schema literal would otherwise recreate `reload`'s identity every render, retriggering - // the effect below in a loop. `latest` still tracks the full schema object so `reload` calls - // the bridge with whatever was passed most recently. const latest = useRef(schema); latest.current = schema; - const [state, setState] = useState({ schemaName: null, status: null, error: null }); + const [state, setState] = useState({ schemaName: null, status: null, error: null }); const reload = useCallback(() => { try { const status = Database.getSyncQueueStatus(latest.current); setState({ schemaName: latest.current.name, status, error: null }); } catch (error) { - // Keep whatever was already on screen — a transient read failure shouldn't blank a known status. setState((prev) => ({ ...prev, error })); } }, [schema.name]); diff --git a/src/hooks/useSyncStatus/types/IState.ts b/src/hooks/useSyncStatus/types/IState.ts new file mode 100644 index 0000000..c792915 --- /dev/null +++ b/src/hooks/useSyncStatus/types/IState.ts @@ -0,0 +1,7 @@ +import type { IUseSyncStatusResult } from './IUseSyncStatusResult'; + +export interface IState { + schemaName: string | null; + status: IUseSyncStatusResult['status']; + error: unknown; +} diff --git a/src/hooks/useSyncStatus/types/index.ts b/src/hooks/useSyncStatus/types/index.ts index 6a9f7c8..02a4b22 100644 --- a/src/hooks/useSyncStatus/types/index.ts +++ b/src/hooks/useSyncStatus/types/index.ts @@ -1 +1,2 @@ export type * from './IUseSyncStatusResult'; +export type * from './IState';