Tracking Cloud Sync Status

let result = try await ditto.store.execute(
  query: "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true AND documents.sync_session_status = 'Connected' ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1"
)
print("Connected to cloud: \(!result.items.isEmpty)")

ditto.store.execute(
    "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true AND documents.sync_session_status = 'Connected' ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
) { result ->
    println("Connected to cloud: ${result.items.isNotEmpty()}")
}

const result = await ditto.store.execute(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true AND documents.sync_session_status = 'Connected' ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1"
);
console.log("Connected to cloud:", result.items.length > 0);

import { Ditto, QueryResult } from "@dittolive/ditto";

const result: QueryResult = await ditto.store.execute(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true AND documents.sync_session_status = 'Connected' ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1"
);
console.log("Connected to cloud:", result.items.length > 0);

final result = await ditto.store.execute(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true AND documents.sync_session_status = 'Connected' ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1"
);
print("Connected to cloud: ${result.items.isNotEmpty}");

try (DittoQueryResult result = ditto.store.execute(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true AND documents.sync_session_status = 'Connected' ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1"
)) {
    System.out.println("Connected to cloud: " + !result.getItems().isEmpty());
}

using var result = await ditto.Store.ExecuteAsync(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true AND documents.sync_session_status = 'Connected' ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1"
);
Console.WriteLine($"Connected to cloud: {result.Items.Count > 0}");

auto result = store.execute(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true AND documents.sync_session_status = 'Connected' ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1"
);
std::cout << "Connected to cloud: " << !result.items().empty() << std::endl;

let result = ditto.store().execute((
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true AND documents.sync_session_status = 'Connected' ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
  serde_json::Value::Null
)).await?;
println!("Connected to cloud: {}", result.item_count() > 0);

result, err := dit.Store().Execute(
    "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true AND documents.sync_session_status = 'Connected' ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1")
if err != nil {
    return err
}
defer result.Close()  // cleanup
fmt.Printf("Connected to cloud: %v\n", result.ItemCount() > 0)

// Step 1: Insert data and get commit ID
let result = try await ditto.store.execute(
  query: "INSERT INTO orders DOCUMENTS (:order)",
  arguments: ["order": ["_id": "order-123", "total": 299.99]]
)
let commitId = result.commitID!

// Step 2: Check if it synced to cloud
let syncStatus = try await ditto.store.execute(
  query: "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1"
)

if let item = syncStatus.items.first,
   let documents = item.value["documents"] as? [String: Any],
   let syncedUpTo = documents["synced_up_to_local_commit_id"] as? Int64 {

  if syncedUpTo >= commitId {
    print("✅ Commit synced to cloud")
  } else {
    print("⏳ Commit pending sync")
  }
}

// Step 1: Insert data and capture its commit ID.
val commitId: ULong = ditto.store.execute(
    query = "INSERT INTO orders DOCUMENTS (:order)",
    arguments = mapOf("order" to mapOf("_id" to "order-123", "total" to 299.99)),
) { result -> result.commitId!! }

// Step 2: Check if it synced to cloud.
ditto.store.execute(
    "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
) { syncStatus ->
    val syncedUpTo = syncStatus.items.firstOrNull()
        ?.value?.get("documents")?.dictionaryOrNull
        ?.let { it["synced_up_to_local_commit_id"].uLongOrNull }
        ?: 0u.toULong()

    if (syncedUpTo >= commitId) {
        println("✅ Commit synced to cloud")
    } else {
        println("⏳ Commit pending sync")
    }
}

// Step 1: Insert data and get commit ID
const result = await ditto.store.execute(
  "INSERT INTO orders DOCUMENTS (:order)",
  { order: { _id: "order-123", total: 299.99 } }
);
const commitId = result.commitID;

// Step 2: Check if it synced to cloud
const syncStatus = await ditto.store.execute(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1"
);

if (syncStatus.items.length > 0) {
  const syncedUpTo = syncStatus.items[0].value.documents.synced_up_to_local_commit_id;

  if (syncedUpTo >= commitId) {
    console.log("✅ Commit synced to cloud");
  } else {
    console.log("⏳ Commit pending sync");
  }
}

import { Ditto, QueryResult } from "@dittolive/ditto";

interface Order {
  _id: string;
  total: number;
}

interface SyncInfoDocuments {
  synced_up_to_local_commit_id: number;
  sync_session_status: string;
}

interface SyncInfoValue {
  documents: SyncInfoDocuments;
}

// Step 1: Insert data and get commit ID
const result: QueryResult = await ditto.store.execute(
  "INSERT INTO orders DOCUMENTS (:order)",
  { order: { _id: "order-123", total: 299.99 } as Order }
);
const commitId: number = result.commitID;

// Step 2: Check if it synced to cloud
const syncStatus: QueryResult = await ditto.store.execute(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1"
);

if (syncStatus.items.length > 0) {
  const value = syncStatus.items[0].value as SyncInfoValue;
  const syncedUpTo: number = value.documents.synced_up_to_local_commit_id;

  if (syncedUpTo >= commitId) {
    console.log("✅ Commit synced to cloud");
  } else {
    console.log("⏳ Commit pending sync");
  }
}

// Step 1: Insert data and get commit ID
final result = await ditto.store.execute(
  "INSERT INTO orders DOCUMENTS (:order)",
  {"order": {"_id": "order-123", "total": 299.99}}
);
final commitId = result.commitID!;

// Step 2: Check if it synced to cloud
final syncStatus = await ditto.store.execute(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1"
);

if (syncStatus.items.isNotEmpty) {
  final documents = syncStatus.items[0].value["documents"] as Map<String, dynamic>;
  final syncedUpTo = documents["synced_up_to_local_commit_id"] as int;

  if (syncedUpTo >= commitId) {
    print("✅ Commit synced to cloud");
  } else {
    print("⏳ Commit pending sync");
  }
}

// Step 1: Insert data and get commit ID
try (DittoQueryResult result = ditto.store.execute(
  "INSERT INTO orders DOCUMENTS (:order)",
  Collections.singletonMap("order",
    Map.of("_id", "order-123", "total", 299.99))
)) {
  long commitId = result.getCommitID();

  // Step 2: Check if it synced to cloud
  try (DittoQueryResult syncStatus = ditto.store.execute(
    "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1"
  )) {
    if (!syncStatus.getItems().isEmpty()) {
      Map<String, Object> value = syncStatus.getItems().get(0).getValue();
      Map<String, Object> documents = (Map<String, Object>) value.get("documents");
      long syncedUpTo = (Long) documents.get("synced_up_to_local_commit_id");

      if (syncedUpTo >= commitId) {
        System.out.println("✅ Commit synced to cloud");
      } else {
        System.out.println("⏳ Commit pending sync");
      }
    }
  }
}

// Step 1: Insert data and get commit ID
using var result = await ditto.Store.ExecuteAsync(
  "INSERT INTO orders DOCUMENTS (:order)",
  new Dictionary<string, object> {
    { "order", new Dictionary<string, object> {
      { "_id", "order-123" },
      { "total", 299.99 }
    }}
  }
);
var commitId = result.CommitID;

// Step 2: Check if it synced to cloud
using var syncStatus = await ditto.Store.ExecuteAsync(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1"
);

if (syncStatus.Items.Count > 0) {
  var value = (Dictionary<string, object>)syncStatus.Items[0].Value;
  var documents = (Dictionary<string, object>)value["documents"];
  var syncedUpTo = (long)documents["synced_up_to_local_commit_id"];

  if (syncedUpTo >= commitId) {
    Console.WriteLine("✅ Commit synced to cloud");
  } else {
    Console.WriteLine("⏳ Commit pending sync");
  }
}

// Step 1: Insert data and get commit ID
auto result = store.execute(
  "INSERT INTO orders DOCUMENTS (:order)",
  {{"order", std::map<std::string, std::any>{
    {"_id", std::string("order-123")},
    {"total", 299.99}
  }}}
);
auto commitId = result.commitID();

// Step 2: Check if it synced to cloud
auto syncStatus = store.execute(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1"
);

if (!syncStatus.items().empty()) {
  const auto& value = syncStatus.items()[0].value();
  const auto& documents = value.at("documents").get<std::map<std::string, std::any>>();
  auto syncedUpTo = std::any_cast<int64_t>(documents.at("synced_up_to_local_commit_id"));

  if (syncedUpTo >= commitId) {
    std::cout << "✅ Commit synced to cloud" << std::endl;
  } else {
    std::cout << "⏳ Commit pending sync" << std::endl;
  }
}

// Step 1: Insert data and get commit ID
let result = ditto.store().execute((
  "INSERT INTO orders DOCUMENTS (:order)",
  serde_json::json!({"order": {"_id": "order-123", "total": 299.99}})
)).await?;
let commit_id = result.commit_id().unwrap();

// Step 2: Check if it synced to cloud
let sync_status = ditto.store().execute((
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
  serde_json::Value::Null
)).await?;

if let Some(item) = sync_status.get_item(0) {
  if let Some(value) = item.value().as_object() {
    if let Some(documents) = value.get("documents").and_then(|d| d.as_object()) {
      if let Some(synced_up_to) = documents.get("synced_up_to_local_commit_id").and_then(|s| s.as_u64()) {
        if synced_up_to >= commit_id {
          println!("✅ Commit synced to cloud");
        } else {
          println!("⏳ Commit pending sync");
        }
      }
    }
  }
}

// Step 1: Insert data and get commit ID
result, err := dit.Store().Execute(
    "INSERT INTO orders DOCUMENTS (:order)",
    ditto.QueryArguments{
        "order" : ditto.Document{
            "_id"   : "order-123",
            "total" : 299.99,
        }
    }
)
if err != nil {
    return err
}
defer result.Close()  // cleanup

commitID, ok := result.CommitID()
if !ok {
    return errors.New("commit ID not available)
}

// Step 2: Check if it synced to Cloud
syncStatus, err := dit.Store().Execute(
    "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1")
if err != nil {
    return err
}
defer syncStatus.Close()  // cleanup

if syncStatus.ItemCount > 0 {
    item := syncStatus.Item(0)
    defer item.Close()  // cleanup
    if documentsValue, ok := item.Value()["documents"]; ok {
        if documentsMap, ok := documentsValue.(map[string]any); ok {
            if syncedUpToValue, ok := documentsMap["synced_up_to_local_commit_id"]; ok {
                if syncedUpTo, ok := syncedUpToValue.(uint64); ok {
                    if syncedUpTo >= commitID {
                        fmt.Println("✅ Commit synced to cloud")
                    } else {
                        fmt.Println("⏳ Commit pending sync"")
                    }
                }
            }
        }
    }
}

let observer = ditto.store.registerObserver(
  query: "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1"
) { result in
  if let item = result.items.first,
     let documents = item.value["documents"] as? [String: Any] {

    let isConnected = documents["sync_session_status"] as? String == "Connected"
    print("Cloud status: \(isConnected ? "Connected" : "Disconnected")")
  } else {
    print("Cloud status: No servers found")
  }
}

// registerObserver auto-closes the DittoQueryResult when the suspend handler
// returns — don't wrap it in `use { }`. Read nested fields via typed
// accessors on the DittoCborSerializable.Dictionary.
val observer = ditto.store.registerObserver(
    "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
) { result ->
    val status = result.items.firstOrNull()
        ?.value?.get("documents")?.dictionaryOrNull
        ?.let { it["sync_session_status"].stringOrNull }
    val isConnected = status == "Connected"
    println("Cloud status: ${if (isConnected) "Connected" else "Disconnected"}")
}

const observer = ditto.store.registerObserver(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
  (result) => {
    if (result.items.length > 0) {
      const isConnected = result.items[0].value.documents.sync_session_status === "Connected";
      console.log("Cloud status:", isConnected ? "Connected" : "Disconnected");
    } else {
      console.log("Cloud status: No servers found");
    }
  }
);

import { Ditto, QueryResult, StoreObserver } from "@dittolive/ditto";

interface SyncInfoDocuments {
  sync_session_status: string;
  synced_up_to_local_commit_id: number;
}

interface SyncInfoValue {
  documents: SyncInfoDocuments;
}

const observer: StoreObserver = ditto.store.registerObserver(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
  (result: QueryResult) => {
    if (result.items.length > 0) {
      const value = result.items[0].value as SyncInfoValue;
      const isConnected: boolean = value.documents.sync_session_status === "Connected";
      console.log("Cloud status:", isConnected ? "Connected" : "Disconnected");
    } else {
      console.log("Cloud status: No servers found");
    }
  }
);

final observer = ditto.store.registerObserver(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
  (result) {
    if (result.items.isNotEmpty) {
      final documents = result.items[0].value["documents"] as Map<String, dynamic>;
      final isConnected = documents["sync_session_status"] == "Connected";
      print("Cloud status: ${isConnected ? "Connected" : "Disconnected"}");
    } else {
      print("Cloud status: No servers found");
    }
  }
);

DittoObserver observer = ditto.store.registerObserver(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
  (result) -> {
    try (result) {
      if (!result.getItems().isEmpty()) {
        Map<String, Object> value = result.getItems().get(0).getValue();
        Map<String, Object> documents = (Map<String, Object>) value.get("documents");
        boolean isConnected = "Connected".equals(documents.get("sync_session_status"));
        System.out.println("Cloud status: " + (isConnected ? "Connected" : "Disconnected"));
      } else {
        System.out.println("Cloud status: No servers found");
      }
    }
  }
);

var observer = ditto.Store.RegisterObserver(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
  (result) => {
    using (result)
    {
      if (result.Items.Count > 0) {
        var value = (Dictionary<string, object>)result.Items[0].Value;
        var documents = (Dictionary<string, object>)value["documents"];
        bool isConnected = (string)documents["sync_session_status"] == "Connected";
        Console.WriteLine($"Cloud status: {(isConnected ? "Connected" : "Disconnected")}");
      } else {
        Console.WriteLine("Cloud status: No servers found");
      }
    }
  }
);

auto observer = ditto.get_store().register_observer(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
  [](const auto& result) {
    if (!result.items().empty()) {
      const auto& value = result.items()[0].value();
      const auto& documents = value.at("documents").get<std::map<std::string, std::any>>();
      bool isConnected = std::any_cast<std::string>(documents.at("sync_session_status")) == "Connected";
      std::cout << "Cloud status: " << (isConnected ? "Connected" : "Disconnected") << std::endl;
    } else {
      std::cout << "Cloud status: No servers found" << std::endl;
    }
  }
);

let observer = ditto.store().register_observer(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
  move |result| {
    if let Some(item) = result.get_item(0) {
      if let Some(value) = item.value().as_object() {
        if let Some(documents) = value.get("documents").and_then(|d| d.as_object()) {
          let is_connected = documents.get("sync_session_status")
            .and_then(|s| s.as_str())
            .map(|s| s == "Connected")
            .unwrap_or(false);
          println!("Cloud status: {}", if is_connected { "Connected" } else { "Disconnected" });
        }
      }
    } else {
      println!("Cloud status: No servers found");
    }
  }
)?;

observer, err := dit.Store().RegisterObserver(
    "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
    nil, // no query arguments
    func(result *ditto.QueryResult) {
        result.Close()  // cleanup

        if result.ItemCount() > 0 {
            item := result.Item(0)
            defer item.Close()  // cleanup

            if documents, ok := item.Value()["documents"].(map[string]any); ok {
                if syncSessionStatus, ok := documents["sync_session_status"].(string); ok {
                    if syncSessionStatus == "Connected" {
                        fmt.Println("Cloud status: Connected")
                    } else {
                        fmt.Println("Cloud status: Disconnected")
                    }
                }
            }
        } else {
            fmt.Println("Cloud status: No servers found")
        }
    }
)
if err != nil {
    return err
}

// Collect the local commit IDs we still need to confirm
var localCommitIds: [Int64] = []

// Watch the cloud sync progress
let observer = ditto.store.registerObserver(
  query: """
    SELECT *
    FROM system:data_sync_info
    WHERE is_ditto_server = true
    ORDER BY documents.synced_up_to_local_commit_id DESC
    LIMIT 1
  """
) { result in
  guard let item = result.items.first,
        let documents = item.value["documents"] as? [String: Any],
        let latestCommitId = documents["synced_up_to_local_commit_id"] as? Int64 else { return }

  // Find the first commitId that hasn't synced yet
  var confirmedCount = 0
  for commitId in localCommitIds {
    if latestCommitId >= commitId {
      print("Successfully synced commit \(commitId) to the cloud")
      confirmedCount += 1
    } else {
      break  // Stop at first non-synced commit
    }
  }

  // Remove synced commitIds
  if confirmedCount > 0 {
    localCommitIds.removeFirst(confirmedCount)
  }
}

// Perform updates and track commit IDs
let result1 = try await ditto.store.execute(query: "UPDATE customers SET verified = true WHERE _id = 'customer-123'")
if let commitId = result1.commitID {
  localCommitIds.append(commitId)
}

let result2 = try await ditto.store.execute(query: "INSERT INTO orders DOCUMENTS ({ _id: 'order-456', total: 599.99 })")
if let commitId = result2.commitID {
  localCommitIds.append(commitId)
}

// Collect the local commit IDs we still need to confirm
val localCommitIds = mutableListOf<ULong>()

// Watch the cloud sync progress. The suspend handler auto-closes the result.
val observer = ditto.store.registerObserver(
    """
    SELECT *
    FROM system:data_sync_info
    WHERE is_ditto_server = true
    ORDER BY documents.synced_up_to_local_commit_id DESC
    LIMIT 1
    """,
) { result ->
    val latestCommitId = result.items.firstOrNull()
        ?.value?.get("documents")?.dictionaryOrNull
        ?.let { it["synced_up_to_local_commit_id"].uLongOrNull }
        ?: return@registerObserver

    var confirmedCount = 0
    for (commitId in localCommitIds) {
        if (latestCommitId >= commitId) {
            println("Successfully synced commit $commitId to the cloud")
            confirmedCount++
        } else {
            break  // stop at first non-synced commit
        }
    }
    if (confirmedCount > 0) {
        repeat(confirmedCount) { localCommitIds.removeAt(0) }
    }
}

// Perform updates and track commit IDs via the handler form of execute.
ditto.store.execute("UPDATE customers SET verified = true WHERE _id = 'customer-123'") { result ->
    result.commitId?.let { localCommitIds.add(it) }
}
ditto.store.execute("INSERT INTO orders DOCUMENTS (:doc)",
    mapOf("doc" to mapOf("_id" to "order-456", "total" to 599.99))) { result ->
    result.commitId?.let { localCommitIds.add(it) }
}

// Collect the local commit IDs we still need to confirm
const localCommitIds = [];

// Watch the cloud sync progress
ditto.store.registerObserver(
  `SELECT *
   FROM system:data_sync_info
   WHERE is_ditto_server = true
   ORDER BY documents.synced_up_to_local_commit_id DESC
   LIMIT 1`,
  (result) => {
    if (!result.items.length) return;

    const latestCommitId = result.items[0].value.documents.synced_up_to_local_commit_id;

    // Find the first commitId that hasn't synced yet
    let confirmedCount = 0;
    for (const commitId of localCommitIds) {
      if (latestCommitId >= commitId) {
        console.log(`Successfully synced commit ${commitId} to the cloud`);
        confirmedCount++;
      } else {
        break;  // Stop at first non-synced commit
      }
    }

    // Modify the array to remove synced commitIds
    if (confirmedCount > 0) {
      localCommitIds.splice(0, confirmedCount);
    }
  }
);

// Perform updates and track commit IDs
const result1 = await ditto.store.execute("UPDATE customers SET verified = true WHERE _id = 'customer-123'");
localCommitIds.push(result1.commitID);

const result2 = await ditto.store.execute("INSERT INTO orders DOCUMENTS ({ _id: 'order-456', total: 599.99 })");
localCommitIds.push(result2.commitID);

import { Ditto, QueryResult, StoreObserver } from "@dittolive/ditto";

interface SyncInfoDocuments {
  synced_up_to_local_commit_id: number;
}

interface SyncInfoValue {
  documents: SyncInfoDocuments;
}

// Collect the local commit IDs we still need to confirm
const localCommitIds: number[] = [];

// Watch the cloud sync progress
const observer: StoreObserver = ditto.store.registerObserver(
  `SELECT *
   FROM system:data_sync_info
   WHERE is_ditto_server = true
   ORDER BY documents.synced_up_to_local_commit_id DESC
   LIMIT 1`,
  (result: QueryResult) => {
    if (!result.items.length) return;

    const value = result.items[0].value as SyncInfoValue;
    const latestCommitId: number = value.documents.synced_up_to_local_commit_id;

    // Find the first commitId that hasn't synced yet
    let confirmedCount: number = 0;
    for (const commitId of localCommitIds) {
      if (latestCommitId >= commitId) {
        console.log(`Successfully synced commit ${commitId} to the cloud`);
        confirmedCount++;
      } else {
        break;  // Stop at first non-synced commit
      }
    }

    // Modify the array to remove synced commitIds
    if (confirmedCount > 0) {
      localCommitIds.splice(0, confirmedCount);
    }
  }
);

// Perform updates and track commit IDs
const result1: QueryResult = await ditto.store.execute("UPDATE customers SET verified = true WHERE _id = 'customer-123'");
localCommitIds.push(result1.commitID);

const result2: QueryResult = await ditto.store.execute("INSERT INTO orders DOCUMENTS ({ _id: 'order-456', total: 599.99 })");
localCommitIds.push(result2.commitID);

// Collect the local commit IDs we still need to confirm
final localCommitIds = <int>[];

// Watch the cloud sync progress
final observer = ditto.store.registerObserver(
  """
    SELECT *
    FROM system:data_sync_info
    WHERE is_ditto_server = true
    ORDER BY documents.synced_up_to_local_commit_id DESC
    LIMIT 1
  """,
  (result) {
    if (result.items.isEmpty) return;

    final documents = result.items[0].value["documents"] as Map<String, dynamic>;
    final latestCommitId = documents["synced_up_to_local_commit_id"] as int;

    // Find the first commitId that hasn't synced yet
    int confirmedCount = 0;
    for (final commitId in localCommitIds) {
      if (latestCommitId >= commitId) {
        print("Successfully synced commit $commitId to the cloud");
        confirmedCount++;
      } else {
        break;  // Stop at first non-synced commit
      }
    }

    // Remove synced commitIds
    if (confirmedCount > 0) {
      localCommitIds.removeRange(0, confirmedCount);
    }
  },
);

// Perform updates and track commit IDs
final result1 = await ditto.store.execute("UPDATE customers SET verified = true WHERE _id = 'customer-123'");
if (result1.commitID != null) {
  localCommitIds.add(result1.commitID!);
}

final result2 = await ditto.store.execute("INSERT INTO orders DOCUMENTS ({ _id: 'order-456', total: 599.99 })");
if (result2.commitID != null) {
  localCommitIds.add(result2.commitID!);
}

// Collect the local commit IDs we still need to confirm
List<Long> localCommitIds = new ArrayList<>();

// Watch the cloud sync progress
DittoObserver observer = ditto.store.registerObserver(
  """
    SELECT *
    FROM system:data_sync_info
    WHERE is_ditto_server = true
    ORDER BY documents.synced_up_to_local_commit_id DESC
    LIMIT 1
  """,
  (result) -> {
    try (result) {
      if (result.getItems().isEmpty()) return;

      Map<String, Object> value = result.getItems().get(0).getValue();
      Map<String, Object> documents = (Map<String, Object>) value.get("documents");
      long latestCommitId = (Long) documents.get("synced_up_to_local_commit_id");

      // Find the first commitId that hasn't synced yet
      int confirmedCount = 0;
      for (Long commitId : localCommitIds) {
        if (latestCommitId >= commitId) {
          System.out.println("Successfully synced commit " + commitId + " to the cloud");
          confirmedCount++;
        } else {
          break;  // Stop at first non-synced commit
        }
      }

      // Remove synced commitIds
      if (confirmedCount > 0) {
        localCommitIds.subList(0, confirmedCount).clear();
      }
    }
  }
);

// Perform updates and track commit IDs
try (DittoQueryResult result1 = ditto.store.execute("UPDATE customers SET verified = true WHERE _id = 'customer-123'")) {
  if (result1.getCommitID() != null) {
    localCommitIds.add(result1.getCommitID());
  }
}


try (DittoQueryResult result2 = ditto.store.execute("INSERT INTO orders DOCUMENTS ({ _id: 'order-456', total: 599.99 })")) {
  if (result2.getCommitID() != null) {
    localCommitIds.add(result2.getCommitID());
  }
}

// Collect the local commit IDs we still need to confirm
List<long> localCommitIds = new List<long>();

// Watch the cloud sync progress
var observer = ditto.Store.RegisterObserver(
  @"
    SELECT *
    FROM system:data_sync_info
    WHERE is_ditto_server = true
    ORDER BY documents.synced_up_to_local_commit_id DESC
    LIMIT 1
  ",
  (result) => {
    using (result) {
      if (result.Items.Count == 0) return;

      var value = (Dictionary<string, object>)result.Items[0].Value;
      var documents = (Dictionary<string, object>)value["documents"];
      var latestCommitId = (long)documents["synced_up_to_local_commit_id"];

      // Find the first commitId that hasn't synced yet
      int confirmedCount = 0;
      foreach (var commitId in localCommitIds) {
        if (latestCommitId >= commitId) {
          Console.WriteLine($"Successfully synced commit {commitId} to the cloud");
          confirmedCount++;
        } else {
          break;  // Stop at first non-synced commit
        }
      }

      // Remove synced commitIds
      if (confirmedCount > 0) {
        localCommitIds.RemoveRange(0, confirmedCount);
      }
    }
  }
);

// Perform updates and track commit IDs
using var result1 = await ditto.Store.ExecuteAsync("UPDATE customers SET verified = true WHERE _id = 'customer-123'");
if (result1.CommitID.HasValue) {
  localCommitIds.Add(result1.CommitID.Value);
}

using var result2 = await ditto.Store.ExecuteAsync("INSERT INTO orders DOCUMENTS ({ _id: 'order-456', total: 599.99 })");
if (result2.CommitID.HasValue) {
  localCommitIds.Add(result2.CommitID.Value);
}

// Collect the local commit IDs we still need to confirm
std::vector<int64_t> localCommitIds;

// Watch the cloud sync progress
auto observer = store.register_observer(
  R"(
    SELECT *
    FROM system:data_sync_info
    WHERE is_ditto_server = true
    ORDER BY documents.synced_up_to_local_commit_id DESC
    LIMIT 1
  )",
  [&localCommitIds](const auto& result) {
    if (result.items().empty()) return;

    const auto& value = result.items()[0].value();
    const auto& documents = value.at("documents").get<std::map<std::string, std::any>>();
    auto latestCommitId = std::any_cast<int64_t>(documents.at("synced_up_to_local_commit_id"));

    // Find the first commitId that hasn't synced yet
    size_t confirmedCount = 0;
    for (const auto& commitId : localCommitIds) {
      if (latestCommitId >= commitId) {
        std::cout << "Successfully synced commit " << commitId << " to the cloud" << std::endl;
        confirmedCount++;
      } else {
        break;  // Stop at first non-synced commit
      }
    }

    // Remove synced commitIds
    if (confirmedCount > 0) {
      localCommitIds.erase(localCommitIds.begin(), localCommitIds.begin() + confirmedCount);
    }
  }
);

// Perform updates and track commit IDs
auto result1 = store.execute("UPDATE customers SET verified = true WHERE _id = 'customer-123'");
if (result1.commitID().has_value()) {
  localCommitIds.push_back(result1.commitID().value());
}

auto result2 = store.execute("INSERT INTO orders DOCUMENTS ({ _id: 'order-456', total: 599.99 })");
if (result2.commitID().has_value()) {
  localCommitIds.push_back(result2.commitID().value());
}

// Collect the local commit IDs we still need to confirm
let mut local_commit_ids: Vec<u64> = Vec::new();

// Watch the cloud sync progress
let observer = ditto.store().register_observer(
  r#"
    SELECT *
    FROM system:data_sync_info
    WHERE is_ditto_server = true
    ORDER BY documents.synced_up_to_local_commit_id DESC
    LIMIT 1
  "#,
  move |result| {
    if result.item_count() == 0 { return; }

    if let Some(item) = result.get_item(0) {
      if let Some(value) = item.value().as_object() {
        if let Some(documents) = value.get("documents").and_then(|d| d.as_object()) {
          if let Some(latest_commit_id) = documents.get("synced_up_to_local_commit_id").and_then(|s| s.as_u64()) {

            // Find the first commitId that hasn't synced yet
            let mut confirmed_count = 0;
            for &commit_id in &local_commit_ids {
              if latest_commit_id >= commit_id {
                println!("Successfully synced commit {} to the cloud", commit_id);
                confirmed_count += 1;
              } else {
                break;  // Stop at first non-synced commit
              }
            }

            // Remove synced commitIds
            if confirmed_count > 0 {
              local_commit_ids.drain(0..confirmed_count);
            }
          }
        }
      }
    }
  }
)?;

// Perform updates and track commit IDs
let result1 = ditto.store().execute(("UPDATE customers SET verified = true WHERE _id = 'customer-123'", serde_json::Value::Null)).await?;
if let Some(commit_id) = result1.commit_id() {
  local_commit_ids.push(commit_id);
}

let result2 = ditto.store().execute(("INSERT INTO orders DOCUMENTS ({ _id: 'order-456', total: 599.99 })", serde_json::Value::Null)).await?;
if let Some(commit_id) = result2.commit_id() {
  local_commit_ids.push(commit_id);
}

// Collect the local commit IDs we still need to confirm
var mu sync.Mutex  // synchronize access to localCommitIDs
var localCommitIDs []uint64

// Watch the cloud sync progress
observer, err := dit.Store().RegisterObserver(
    `SELECT *
    FROM system:data_sync_info
    WHERE is_ditto_server = true
    ORDER BY documents.synced_up_to_local_commit_id DESC
    LIMIT 1`,
    nil, // no query arguments
    func(result *ditto.QueryResult) {
        defer result.Close()  // cleanup

        if result.ItemCount() == 0 {
            return
        }

        item := result.Item(0)
        defer item.Close()  // cleanup
        if documents, ok := item.Value()["documents"].(map[string]any); ok {
            if latestCommitID, ok := documents["synced_up_to_local_commit_id].(uint64); ok {
                mu.Lock()
                defer mu.Unlock()

                // Find the first commit ID that hasn't synced
                confirmedCount := 0
                for _, commitId := range localCommitIDs {
                    if latestCommitID >= commitID {
                        fmt.Printf("Successfully synced commit %u to the cloud", commitID)
                        confirmedCount++
                    } else {
                        break  // stop at first non-synced commit
                    }
                }

                // Remove synced commit IDs
                localCommitIDs :+ localCommitIDs[confirmedCount:]
            }
        }
    }
)
if err != nil {
    return err
}

// Perform updates and track commit IDs
result1, err := dit.Store().Execute("UPDATE customers SET verified = true WHERE _id = 'customer-123'")
if err != nil {
    return err
}
defer result1.Close()  // cleanup
if commitID, ok := result1.CommitID(); ok {
    mu.Lock()
    defer mu.Unlock()
    localCommitIDs := append(localCommitIDs, commitID)
}

result2, err := dit.Store().Execute("INSERT INTO orders DOCUMENTS ({ _id: 'order-456', total: 599.99 })")
if err != nil {
    return err
}
defer result2.Close()  // cleanup
if commitID, ok := result2.CommitID(); ok {
    mu.Lock()
    defer mu.Unlock()
    localCommitIDs := append(localCommitIDs, commitID)
}

// Watch for incoming data from cloud
let observer = ditto.store.registerObserver(
  query: "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1"
) { result in
  guard let item = result.items.first,
        let documents = item.value["documents"] as? [String: Any],
        let lastUpdate = documents["last_update_received_time"] as? Int64 else { return }

  if lastUpdate > 0 {
    print("📥 New data received from cloud at: \(Date(timeIntervalSince1970: TimeInterval(lastUpdate)))")
  }
}

// Watch for incoming data from cloud. The handler auto-closes the result.
val observer = ditto.store.registerObserver(
    "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
) { result ->
    val lastUpdate = result.items.firstOrNull()
        ?.value?.get("documents")?.dictionaryOrNull
        ?.let { it["last_update_received_time"].longOrNull }
        ?: return@registerObserver

    if (lastUpdate > 0) {
        println("📥 New data received from cloud at: ${java.time.Instant.ofEpochSecond(lastUpdate)}")
    }
}

// Watch for incoming data from cloud
ditto.store.registerObserver(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
  (result) => {
    if (!result.items.length) return;

    const lastUpdate = result.items[0].value.documents.last_update_received_time;

    if (lastUpdate > 0) {
      console.log("📥 New data received from cloud at:", new Date(lastUpdate * 1000));
    }
  }
);

import { Ditto, QueryResult, StoreObserver } from "@dittolive/ditto";

interface SyncInfoDocuments {
  last_update_received_time: number;
  synced_up_to_local_commit_id: number;
}

interface SyncInfoValue {
  documents: SyncInfoDocuments;
}

// Watch for incoming data from cloud
const observer: StoreObserver = ditto.store.registerObserver(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
  (result: QueryResult) => {
    if (!result.items.length) return;

    const value = result.items[0].value as SyncInfoValue;
    const lastUpdate: number = value.documents.last_update_received_time;

    if (lastUpdate > 0) {
      console.log("📥 New data received from cloud at:", new Date(lastUpdate * 1000));
    }
  }
);

// Watch for incoming data from cloud
final observer = ditto.store.registerObserver(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
  (result) {
    if (result.items.isEmpty) return;

    final documents = result.items[0].value["documents"] as Map<String, dynamic>;
    final lastUpdate = documents["last_update_received_time"] as int;

    if (lastUpdate > 0) {
      print("📥 New data received from cloud at: ${DateTime.fromMillisecondsSinceEpoch(lastUpdate * 1000)}");
    }
  },
);

// Watch for incoming data from cloud
DittoObserver observer = ditto.store.registerObserver(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
  (result) -> {
    try (result)
    {
      if (result.getItems().isEmpty()) return;

      Map<String, Object> value = result.getItems().get(0).getValue();
      Map<String, Object> documents = (Map<String, Object>) value.get("documents");
      long lastUpdate = (Long) documents.get("last_update_received_time");

      if (lastUpdate > 0) {
        System.out.println("📥 New data received from cloud at: " + java.time.Instant.ofEpochSecond(lastUpdate));
      }
    }
  }
);

// Watch for incoming data from cloud
var observer = ditto.Store.RegisterObserver(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
  (result) => {
    using (result)
    {
      if (result.Items.Count == 0) return;

      var value = (Dictionary<string, object>)result.Items[0].Value;
      var documents = (Dictionary<string, object>)value["documents"];
      var lastUpdate = (long)documents["last_update_received_time"];

      if (lastUpdate > 0) {
        Console.WriteLine($"📥 New data received from cloud at: {DateTimeOffset.FromUnixTimeSeconds(lastUpdate)}");
      }
    }
  }
);

// Watch for incoming data from cloud
auto observer = store.register_observer(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
  [](const auto& result) {
    if (result.items().empty()) return;

    const auto& value = result.items()[0].value();
    const auto& documents = value.at("documents").get<std::map<std::string, std::any>>();
    auto lastUpdate = std::any_cast<int64_t>(documents.at("last_update_received_time"));

    if (lastUpdate > 0) {
      std::cout << "📥 New data received from cloud at: " << lastUpdate << std::endl;
    }
  }
);

// Watch for incoming data from cloud
let observer = ditto.store().register_observer(
  "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
  move |result| {
    if result.item_count() == 0 { return; }

    if let Some(item) = result.get_item(0) {
      if let Some(value) = item.value().as_object() {
        if let Some(documents) = value.get("documents").and_then(|d| d.as_object()) {
          if let Some(last_update) = documents.get("last_update_received_time").and_then(|s| s.as_i64()) {

            if last_update > 0 {
              println!("📥 New data received from cloud at: {}", last_update);
            }
          }
        }
      }
    }
  }
)?;

// Watch for incoming data from cloud
observer, err := dit.Store().RegisterObserver(
    "SELECT * FROM system:data_sync_info WHERE is_ditto_server = true ORDER BY documents.synced_up_to_local_commit_id DESC LIMIT 1",
    nil,  // no query parameters
    func(result *ditto.QueryResult) {
        defer result.Close()  // cleanup

        if result.ItemCount == 0 {
            return
        }

        item := result.Item(0)
        defer item.Close()  // cleanup
        if documents, ok := item.Value()["documents"].(map[string]any); ok {
            if lastUpdate, ok := documents["last_update_received_time].(int64); ok {
                if lastUpdate > 0 {
                    fmt.Printf("📥 New data received from cloud at: %d\n", lastUpdate)
                }
            }
        }
    }
)
if err != nil {
    return err
}