Sync time with ntp server, starting with basic operations

1 files changed, 106 insertions, 7 deletions

diff --git a/src/Database.cpp b/src/Database.cpp
index 4fa8405..33aef54 100644
--- a/src/Database.cpp
+++ b/src/Database.cpp
@@ -1,9 +1,22 @@
 #include "../include/Database.hpp"
+#include "../include/Group.hpp"
+#include "../include/User.hpp"
 #include <opendht.h>
 #include <fmt/format.h>
+#include <sodium/crypto_box_curve25519xchacha20poly1305.h>
+#include <thread>
+#include <chrono>
+#include <sibs/SafeSerializer.hpp>
 
 using namespace dht;
 using namespace std;
+using namespace chrono_literals;
+
+static int databaseCount = 0;
+// TODO: Verify time_t is always signed
+static time_t timeOffset = 0; // Updated by comparing local time with ntp server
+static thread *ntpThread = nullptr;
+static bool timestampSynced = false;
 
 namespace odhtdb
 {
@@ -13,36 +26,122 @@ namespace odhtdb
         fmt::MemoryWriter portStr;
         portStr << port;
         node.bootstrap(bootstrapNodeAddr, portStr.c_str());
+
+        // TODO: Make this work for multiple threads initializing database at same time
+        ++databaseCount;
+        if(databaseCount == 1)
+        {
+            if(ntpThread)
+                delete ntpThread;
+
+            ntpThread = new thread([]()
+            {
+                ntp::NtpClient ntpClient("pool.ntp.org");
+                while(databaseCount > 0)
+                {
+                    ntp::NtpTimestamp ntpTimestamp = ntpClient.getTimestamp();
+                    timeOffset = time(nullptr) - ntpTimestamp.seconds;
+                    timestampSynced = true;
+                    // TODO: Also use timestamp fraction (milliseconds)
+                    this_thread::sleep_for(60s);
+                }
+                timestampSynced = false;
+            });
+
+            // TODO: Catch std::system_error instead of this if-statement
+            if(ntpThread->joinable())
+                ntpThread->detach();
+        }
+
+        while(!timestampSynced)
+        {
+            this_thread::sleep_for(10ms);
+        }
     }
 
     Database::~Database()
     {
+        // TODO: Make this work for multiple threads removing database object at same time
+        --databaseCount;
         node.join();
     }
 
+    void Database::create(const Key &key, Group *primaryAdminGroup)
+    {
+        // TODO: Append fractions to get real microseconds time
+        u64 timeMicroseconds = ((u64)getSyncedTimestampUtc().seconds) * 1000000ull;
+        stagedCreateObjects.emplace_back(StagedCreateObject{ key, primaryAdminGroup, timeMicroseconds });
+    }
+
     void Database::add(const Key &key, DataView data)
     {
-        stagedObjects.emplace_back(StagedObject{ key, data });
+        // TODO: Append fractions to get real microseconds time
+        u64 timeMicroseconds = ((u64)getSyncedTimestampUtc().seconds) * 1000000ull;
+        stagedAddObjects.emplace_back(StagedAddObject{ key, data, timeMicroseconds });
     }
 
     void Database::commit()
     {
-        // TODO: Combine staged objects into one object
-        for(StagedObject stagedObject : stagedObjects)
+        // TODO: Combine staged objects into one object for efficiency.
+        // TODO: Add rollback
+
+        for(StagedCreateObject &stagedObject : stagedCreateObjects)
         {
-            commitStagedObject(stagedObject);
+            commitStagedCreateObject(stagedObject);
         }
-        stagedObjects.clear();
+        stagedCreateObjects.clear();
+
+        for(StagedAddObject &stagedObject : stagedAddObjects)
+        {
+            commitStagedAddObject(stagedObject);
+        }
+        stagedAddObjects.clear();
+    }
+
+    void Database::commitStagedCreateObject(const StagedCreateObject &stagedObject)
+    {
+        // TODO: Use (ed25519 or poly1305) and curve25519
+        // TODO: Implement gas and price (refill when serving content (seeding) or by waiting. This is done to prevent spamming and bandwidth leeching)
+        sibs::SafeSerializer serializer;
+        serializer.add(stagedObject.key.hashedKey.data(), stagedObject.key.hashedKey.size());
+        serializer.add((u8)stagedObject.primaryAdminGroup->getName().size());
+        serializer.add((u8*)stagedObject.primaryAdminGroup->getName().data(), stagedObject.primaryAdminGroup->getName().size());
+        serializer.add((u32)stagedObject.primaryAdminGroup->getUsers().size());
+        for(User *user : stagedObject.primaryAdminGroup->getUsers())
+        {
+            serializer.add((u8)user->getName().size());
+            serializer.add((u8*)user->getName().data(), user->getName().size());
+        }
+        
+        // TODO: Verify if serializer buffer needs to survive longer than this method
+        Value value(serializer.getBuffer().data(), serializer.getBuffer().size());
+        node.put(stagedObject.key.hashedKey, move(value), [](bool ok)
+        {
+            // TODO: Handle failure to put data
+            if(!ok)
+                fprintf(stderr, "Failed to put: %s, what to do?\n", "commitStagedCreateObject");
+        }, time_point(), false);
     }
 
-    void Database::commitStagedObject(const StagedObject &stagedObject)
+    void Database::commitStagedAddObject(const StagedAddObject &stagedObject)
     {
+        // TODO: Use (ed25519 or poly1305) and curve25519
+        // TODO: Implement gas and price (refill when serving content (seeding) or by waiting. This is done to prevent spamming and bandwidth leeching)
         Value value((const u8*)stagedObject.data.data, stagedObject.data.size);
         node.put(stagedObject.key.hashedKey, move(value), [](bool ok)
         {
             // TODO: Handle failure to put data
             if(!ok)
-                fprintf(stderr, "Failed to put: %s, what to do?\n", "commitStagedObject");
+                fprintf(stderr, "Failed to put: %s, what to do?\n", "commitStagedAddObject");
         }, time_point(), false);
     }
+
+    ntp::NtpTimestamp Database::getSyncedTimestampUtc() const
+    {
+        assert(timestampSynced);
+        ntp::NtpTimestamp timestamp;
+        timestamp.seconds = time(nullptr) - timeOffset;
+        timestamp.fractions = 0; // TODO: Set this
+        return timestamp;
+    }
 }
\ No newline at end of file