1
0
This repository has been archived on 2025-07-27. You can view files and clone it, but cannot push or open issues or pull requests.
Files
siadrive/src/siadrive_api/uploadmanager.cpp
Scott E. Graves 0a04147eb2 Fix database error
2017-04-17 23:57:12 -05:00

1169 lines
29 KiB
C++

#include <SQLiteCpp/Exception.h>
#include <uploadmanager.h>
#include <siaapi.h>
#include <eventsystem.h>
#include <siadriveconfig.h>
#include <filepath.h>
using namespace Sia::Api;
// Event Notifications
class UploadManagerStarted :
public CEvent
{
public:
UploadManagerStarted()
{
}
public:
virtual ~UploadManagerStarted()
{
}
public:
virtual SString GetSingleLineMessage() const override
{
return GetEventName();
}
virtual std::shared_ptr<CEvent> Clone() const override
{
return std::shared_ptr<CEvent>(new UploadManagerStarted());
}
virtual SString GetEventName() const override
{
return "UploadManagerStarted";
}
virtual json GetEventJson() const override
{
return
{
{ "event", GetEventName() }
};
}
};
class UploadManagerStopped :
public CEvent
{
public:
UploadManagerStopped()
{
}
public:
virtual ~UploadManagerStopped()
{
}
public:
virtual SString GetSingleLineMessage() const override
{
return GetEventName();
}
virtual std::shared_ptr<CEvent> Clone() const override
{
return std::shared_ptr<CEvent>(new UploadManagerStopped());
}
virtual SString GetEventName() const override
{
return "UploadManagerStopped";
}
virtual json GetEventJson() const override
{
return
{
{ "event", GetEventName() }
};
}
};
class FileAddedToQueue :
public CEvent
{
public:
FileAddedToQueue(const SString& siaPath, const SString& filePath) :
_siaPath(siaPath),
_filePath(filePath)
{
}
public:
virtual ~FileAddedToQueue()
{
}
private:
const SString _siaPath;
const SString _filePath;
public:
virtual SString GetSingleLineMessage() const override
{
return GetEventName() +
"|SP|" + _siaPath +
"|FP|" + _filePath;
}
virtual std::shared_ptr<CEvent> Clone() const override
{
return std::shared_ptr<CEvent>(new FileAddedToQueue(_siaPath, _filePath));
}
virtual SString GetEventName() const override
{
return "FileAddedToQueue";
}
virtual json GetEventJson() const override
{
return
{
{ "event", GetEventName() },
{ "sia_path", _siaPath },
{ "file_path", _filePath }
};
}
};
class ExternallyRemovedFileDetected :
public CEvent
{
public:
ExternallyRemovedFileDetected(const SString& siaPath, const SString& filePath) :
_siaPath(siaPath),
_filePath(filePath)
{
}
public:
virtual ~ExternallyRemovedFileDetected()
{
}
private:
const SString _siaPath;
const SString _filePath;
public:
virtual SString GetSingleLineMessage() const override
{
return GetEventName() +
"|SP|" + _siaPath +
"|FP|" + _filePath;
}
virtual std::shared_ptr<CEvent> Clone() const override
{
return std::shared_ptr<CEvent>(new ExternallyRemovedFileDetected(_siaPath, _filePath));
}
virtual SString GetEventName() const override
{
return "ExternallyRemovedFileDetected";
}
virtual json GetEventJson() const override
{
return
{
{ "event", GetEventName() },
{ "sia_path", _siaPath },
{ "file_path", _filePath }
};
}
};
class UploadToSiaStarted :
public CEvent
{
public:
UploadToSiaStarted(const SString& siaPath, const SString& filePath) :
_siaPath(siaPath),
_filePath(filePath)
{
}
public:
virtual ~UploadToSiaStarted()
{
}
private:
const SString _siaPath;
const SString _filePath;
public:
virtual SString GetSingleLineMessage() const override
{
return GetEventName() +
"|SP|" + _siaPath +
"|FP|" + _filePath;
}
virtual std::shared_ptr<CEvent> Clone() const override
{
return std::shared_ptr<CEvent>(new UploadToSiaStarted(_siaPath, _filePath));
}
virtual SString GetEventName() const override
{
return "UploadToSiaStarted";
}
virtual json GetEventJson() const override
{
return
{
{ "event", GetEventName() },
{ "sia_path", _siaPath },
{ "file_path", _filePath }
};
}
};
class UploadToSiaComplete :
public CEvent
{
public:
UploadToSiaComplete(const SString& siaPath, const SString& filePath) :
_siaPath(siaPath),
_filePath(filePath)
{
}
public:
virtual ~UploadToSiaComplete()
{
}
private:
const SString _siaPath;
const SString _filePath;
public:
virtual SString GetSingleLineMessage() const override
{
return GetEventName() +
"|SP|" + _siaPath +
"|FP|" + _filePath;
}
virtual std::shared_ptr<CEvent> Clone() const override
{
return std::shared_ptr<CEvent>(new UploadToSiaComplete(_siaPath, _filePath));
}
virtual SString GetEventName() const override
{
return "UploadToSiaComplete";
}
virtual json GetEventJson() const override
{
return
{
{ "event", GetEventName() },
{ "sia_path", _siaPath },
{ "file_path", _filePath }
};
}
};
class FileRemovedFromSia :
public CEvent
{
public:
FileRemovedFromSia(const SString& siaPath, const SString& filePath) :
_siaPath(siaPath),
_filePath(filePath)
{
}
public:
virtual ~FileRemovedFromSia()
{
}
private:
const SString _siaPath;
const SString _filePath;
public:
virtual SString GetSingleLineMessage() const override
{
return GetEventName() +
"|SP|" + _siaPath +
"|FP|" + _filePath;
}
virtual std::shared_ptr<CEvent> Clone() const override
{
return std::shared_ptr<CEvent>(new FileRemovedFromSia(_siaPath, _filePath));
}
virtual SString GetEventName() const override
{
return "FileRemovedFromSia";
}
virtual json GetEventJson() const override
{
return
{
{ "event", GetEventName() },
{ "sia_path", _siaPath },
{ "file_path", _filePath }
};
}
};
class FileRemovedFromDatabase :
public CEvent
{
public:
FileRemovedFromDatabase(const SString& siaPath, const SString& filePath) :
_siaPath(siaPath),
_filePath(filePath)
{
}
public:
virtual ~FileRemovedFromDatabase()
{
}
private:
const SString _siaPath;
const SString _filePath;
public:
virtual SString GetSingleLineMessage() const override
{
return GetEventName() +
"|SP|" + _siaPath +
"|FP|" + _filePath;
}
virtual std::shared_ptr<CEvent> Clone() const override
{
return std::shared_ptr<CEvent>(new FileRemovedFromDatabase(_siaPath, _filePath));
}
virtual SString GetEventName() const override
{
return "FileRemovedFromDatabase";
}
virtual json GetEventJson() const override
{
return
{
{ "event", GetEventName() },
{ "sia_path", _siaPath },
{ "file_path", _filePath }
};
}
};
class FailedToRemoveFileFromSia :
public CEvent
{
public:
FailedToRemoveFileFromSia(const SString& siaPath, const SString& filePath, const SiaCurlError& curlError) :
_siaPath(siaPath),
_filePath(filePath),
_curlError(curlError)
{
}
public:
virtual ~FailedToRemoveFileFromSia()
{
}
private:
const SString _siaPath;
const SString _filePath;
const SiaCurlError _curlError;
public:
virtual SString GetSingleLineMessage() const override
{
return GetEventName() +
"|SP|" + _siaPath +
"|FP|" + _filePath +
"|CERR|" + _curlError.GetReason();
}
virtual std::shared_ptr<CEvent> Clone() const override
{
return std::shared_ptr<CEvent>(new FailedToRemoveFileFromSia(_siaPath, _filePath, _curlError));
}
virtual SString GetEventName() const override
{
return "FailedToRemoveFileFromSia";
}
virtual json GetEventJson() const override
{
return
{
{ "event", GetEventName() },
{ "sia_path", _siaPath },
{ "file_path", _filePath },
{ "curl_error", _curlError.GetReason() }
};
}
};
class ModifyUploadStatusFailed :
public CEvent
{
public:
ModifyUploadStatusFailed(const SString& siaPath, const SString& filePath, const UploadStatus& uploadStatus, const SString& errorMsg) :
_siaPath(siaPath),
_filePath(filePath),
_uploadStatus(uploadStatus),
_errorMsg(errorMsg)
{
}
public:
virtual ~ModifyUploadStatusFailed()
{
}
private:
const SString _siaPath;
const SString _filePath;
const UploadStatus _uploadStatus;
const SString _errorMsg;
public:
virtual SString GetSingleLineMessage() const override
{
return GetEventName() +
"|SP|" + _siaPath +
"|FP|" + _filePath +
"|ST|" + CUploadManager::UploadStatusToString(_uploadStatus) +
"|MSG|" + _errorMsg;
}
virtual std::shared_ptr<CEvent> Clone() const override
{
return std::shared_ptr<CEvent>(new ModifyUploadStatusFailed(_siaPath, _filePath, _uploadStatus, _errorMsg));
}
virtual SString GetEventName() const override
{
return "ModifyUploadStatusFailed";
}
virtual json GetEventJson() const override
{
return
{
{ "event", GetEventName() },
{ "sia_path", _siaPath },
{ "file_path", _filePath },
{ "status", CUploadManager::UploadStatusToString(_uploadStatus) },
{ "message", _errorMsg}
};
}
};
class DatabaseInsertFailed :
public CEvent
{
public:
DatabaseInsertFailed(const SString& siaPath, const SString& filePath, const SString& errorMessage) :
_siaPath(siaPath),
_filePath(filePath),
_errorMsg(errorMessage)
{
}
public:
virtual ~DatabaseInsertFailed()
{
}
private:
const SString _siaPath;
const SString _filePath;
const SString _errorMsg;
public:
virtual SString GetSingleLineMessage() const override
{
return GetEventName() +
"|SP|" + _siaPath +
"|FP|" + _filePath +
"|MSG|" + _errorMsg;
}
virtual std::shared_ptr<CEvent> Clone() const override
{
return std::shared_ptr<CEvent>(new DatabaseInsertFailed(_siaPath, _filePath, _errorMsg));
}
virtual SString GetEventName() const override
{
return "DatabaseInsertFailed";
}
virtual json GetEventJson() const override
{
return
{
{ "event", GetEventName() },
{ "sia_path", _siaPath },
{ "file_path", _filePath },
{ "message", _errorMsg }
};
}
};
class DatabaseDeleteFailed :
public CEvent
{
public:
DatabaseDeleteFailed(const SString& siaPath, const SString& filePath, const SString& errorMessage) :
_siaPath(siaPath),
_filePath(filePath),
_errorMsg(errorMessage)
{
}
public:
virtual ~DatabaseDeleteFailed()
{
}
private:
const SString _siaPath;
const SString _filePath;
const SString _errorMsg;
public:
virtual SString GetSingleLineMessage() const override
{
return GetEventName() +
"|SP|" + _siaPath +
"|FP|" + _filePath +
"|MSG|" + _errorMsg;
}
virtual std::shared_ptr<CEvent> Clone() const override
{
return std::shared_ptr<CEvent>(new DatabaseDeleteFailed(_siaPath, _filePath, _errorMsg));
}
virtual SString GetEventName() const override
{
return "DatabaseDeleteFailed";
}
virtual json GetEventJson() const override
{
return
{
{ "event", GetEventName() },
{ "sia_path", _siaPath },
{ "file_path", _filePath },
{ "message", _errorMsg }
};
}
};
class SourceFileNotFound :
public CEvent
{
public:
SourceFileNotFound(const SString& siaPath, const SString& filePath) :
_siaPath(siaPath),
_filePath(filePath)
{
}
public:
virtual ~SourceFileNotFound()
{
}
private:
const SString _siaPath;
const SString _filePath;
public:
virtual std::shared_ptr<CEvent> Clone() const override
{
return std::shared_ptr<CEvent>(new SourceFileNotFound(_siaPath, _filePath));
}
virtual SString GetSingleLineMessage() const override
{
return GetEventName() +
"|SP|" + _siaPath +
"|FP|" + _filePath;
}
virtual SString GetEventName() const override
{
return "SourceFileNotFound";
}
virtual json GetEventJson() const override
{
return
{
{ "event", GetEventName() },
{ "sia_path", _siaPath },
{ "file_path", _filePath }
};
}
};
class DatabaseExceptionOccurred :
public CEvent
{
public:
DatabaseExceptionOccurred(const SString& duringOperation, const SQLite::Exception& exception) :
_duringOperation(duringOperation),
_exception(exception)
{
}
public:
virtual ~DatabaseExceptionOccurred()
{
}
private:
const SString _duringOperation;
const SQLite::Exception _exception;
public:
virtual std::shared_ptr<CEvent> Clone() const override
{
return std::shared_ptr<CEvent>(new DatabaseExceptionOccurred(_duringOperation, _exception));
}
virtual SString GetSingleLineMessage() const override
{
return GetEventName() +
"|MSG|" + SString(_exception.what()) +
"|OPR|" + _duringOperation;
}
virtual SString GetEventName() const override
{
return "DatabaseExceptionOccurred";
}
virtual json GetEventJson() const override
{
return
{
{ "event", GetEventName() },
{ "message", _exception.getErrorStr() },
{ "operation", _duringOperation }
};
}
};
#define TABLE_CREATE L"create table if not exists %s (%s);"
#define UPLOAD_TABLE L"upload_table"
#define UPLOAD_TABLE_COLUMNS L"id integer primary key autoincrement, sia_path text unique not null, file_path text unique not null, status integer not null, modified_time integer not null"
#define QUERY_STATUS "select * from upload_table where sia_path=@sia_path order by id desc limit 1;"
#define QUERY_UPLOADS "select * from upload_table order by sia_path desc;"
#define QUERY_UPLOADS_BY_STATUS "select * from upload_table where status=@status order by id desc limit 1;"
#define QUERY_UPLOADS_BY_2_STATUS "select * from upload_table where status=@status1 or status=@status2 order by sia_path desc;"
#define QUERY_UPLOAD_COUNT_BY_STATUS "select count(id) from upload_table where status=@status;"
#define QUERY_UPLOADS_BY_SIA_PATH "select * from upload_table where sia_path=@sia_path order by id desc limit 1;"
#define QUERY_UPLOADS_BY_SIA_PATH_AND_STATUS "select * from upload_table where sia_path=@sia_path and status=@status order by id desc limit 1;"
#define UPDATE_STATUS "update upload_table set status=@status where sia_path=@sia_path;"
#define INSERT_UPLOAD "insert into upload_table (sia_path, status, file_path, modified_time) values (@sia_path, @status, @file_path, @modified_time);"
#define DELETE_UPLOAD "delete from upload_table where sia_path=@sia_path;"
#define SET_STATUS(status, success_event, fail_event)\
bool statusUpdated = false;\
try\
{\
SQLite::Statement update(_uploadDatabase, UPDATE_STATUS);\
update.bind("@sia_path", SString::ToUtf8(siaPath).c_str());\
update.bind("@status", static_cast<unsigned>(status));\
if (update.exec() != 1)\
{\
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(fail_event(siaPath, filePath, status, update.getErrorMsg())));\
}\
else\
{\
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(success_event(siaPath, filePath)));\
statusUpdated = true;\
}\
}\
catch (SQLite::Exception e)\
{\
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(DatabaseExceptionOccurred("SetStatus", e)));\
}
template <typename... Ts>
SString fmt(const SString &fmt, Ts... vs)
{
size_t required = _snwprintf(nullptr, 0, fmt.str().c_str(), vs...);
SString ret;
ret.Resize(required);
_snwprintf(&ret[0], required, fmt.str().c_str(), vs...);
return ret;
}
static void CreateTableIfNotFound(SQLite::Database* database, const SString& tableName, const SString& columns)
{
SString sqlCreate = fmt(TABLE_CREATE, &tableName[0], &columns[0]);
database->exec(SString::ToUtf8(sqlCreate).c_str());
}
SString CUploadManager::UploadStatusToString(const UploadStatus& uploadStatus)
{
switch (uploadStatus)
{
case UploadStatus::Complete:
return L"Complete";
case UploadStatus::NotFound:
return L"Not Found";
case UploadStatus::Queued:
return L"Queued";
case UploadStatus::Uploading:
return L"Uploading";
default:
return L"!!Not Defined!!";
}
}
CUploadManager::CUploadManager(const CSiaCurl& siaCurl, CSiaDriveConfig* siaDriveConfig) :
CAutoThread(siaCurl, siaDriveConfig),
_siaDriveConfig(siaDriveConfig),
_uploadDatabase(siaDriveConfig->GetRenter_UploadDbFilePath(), SQLite::OPEN_CREATE | SQLite::OPEN_READWRITE, 5000)
{
CreateTableIfNotFound(&_uploadDatabase, UPLOAD_TABLE, UPLOAD_TABLE_COLUMNS);
// Detect files that have been removed since last startup
DeleteFilesNotFound(siaCurl, siaDriveConfig);
DeleteFilesRemovedFromSia(siaCurl, siaDriveConfig, true);
// Begin normal processing
StartAutoThread();
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(UploadManagerStarted()));
}
CUploadManager::~CUploadManager()
{
// Stop all processing
StopAutoThread();
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(UploadManagerStopped()));
}
void CUploadManager::DeleteFilesNotFound(const CSiaCurl& siaCurl, CSiaDriveConfig* siaDriveConfig)
{
std::vector<std::pair<SString, SString>> removeList;
{
SQLite::Statement query(_uploadDatabase, QUERY_UPLOADS);
while (query.executeStep())
{
SString filePath = static_cast<const char*>(query.getColumn(query.getColumnIndex("file_path")));
if (!FilePath(filePath).IsFile())
{
SString siaPath = static_cast<const char*>(query.getColumn(query.getColumnIndex("sia_path")));
removeList.push_back({ siaPath, filePath });
}
}
}
for (const auto& item : removeList)
{
SString siaPath = item.first;
SString filePath = item.second;
SQLite::Statement del(_uploadDatabase, DELETE_UPLOAD);
del.bind("@sia_path", SString::ToUtf8(siaPath).c_str());
if (del.exec() >= 0)
{
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(FileRemovedFromDatabase(siaPath, filePath)));
}
else
{
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(DatabaseDeleteFailed(siaPath, filePath, del.getErrorMsg())));
}
}
}
void CUploadManager::DeleteFilesRemovedFromSia(const CSiaCurl& siaCurl, CSiaDriveConfig* siaDriveConfig, const bool& isStartup)
{
CSiaFileTreePtr fileTree(new CSiaFileTree(siaCurl, siaDriveConfig));
json result;
SiaCurlError cerror = siaCurl.Get(L"/renter/files", result);
if (ApiSuccess(cerror))
{
fileTree->BuildTree(result);
auto fileList = fileTree->GetFileList();
// TODO Implement this
}
else if (isStartup)
{
throw StartupException(L"Failed to get Sia files");
}
}
void CUploadManager::HandleFileRemove(const CSiaCurl& siaCurl, const SString& siaPath)
{
FilePath removeFilePath(GetSiaDriveConfig()->GetCacheFolder(), siaPath);
json response;
SiaCurlError cerror = siaCurl.Post(SString(L"/renter/delete/") + siaPath, {}, response);
if (ApiSuccess(cerror) || cerror.GetReason().Contains("no file known"))
{
SQLite::Statement del(_uploadDatabase, DELETE_UPLOAD);
del.bind("@sia_path", SString::ToUtf8(siaPath).c_str());
auto delCount = del.exec();
if (delCount >= 0)
{
if (delCount != 0)
{
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(FileRemovedFromSia(siaPath, removeFilePath)));
}
}
else
{
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(DatabaseDeleteFailed(siaPath, removeFilePath, del.getErrorMsg())));
}
}
else
{
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(FailedToRemoveFileFromSia(siaPath, removeFilePath, cerror)));
CEventSystem::EventSystem.NotifyEvent(CreateSystemCriticalEvent(cerror.GetReason()));
}
}
void CUploadManager::AutoThreadCallback(const CSiaCurl& siaCurl, CSiaDriveConfig* siaDriveConfig)
{
bool processNext = true;
do
{
try
{
CSiaFileTreePtr fileTree(new CSiaFileTree(siaCurl, siaDriveConfig));
json result;
if (ApiSuccess(siaCurl.Get(L"/renter/files", result)))
{
fileTree->BuildTree(result);
try
{
UploadFileListPtr uploadFileList(new UploadFileList());
std::lock_guard<std::mutex> l(_uploadMutex);
SQLite::Statement query(_uploadDatabase, QUERY_UPLOADS_BY_2_STATUS);
query.bind("@status1", static_cast<unsigned>(UploadStatus::Queued));
query.bind("@status2", static_cast<unsigned>(UploadStatus::Uploading));
while (query.executeStep())
{
SString siaPath = static_cast<const char*>(query.getColumn(query.getColumnIndex("sia_path")));
auto fileList = fileTree->GetFileList();
auto it = std::find_if(fileList->begin(), fileList->end(), [&](const CSiaFilePtr& ptr)
{
return ptr->GetSiaPath() == siaPath;
});
if (it == fileList->end())
{
// TODO Fix file size
CSiaFilePtr ptr(new CSiaFile(siaCurl, siaDriveConfig,
{
{"siapath", siaPath},
{"filesize", 0},
{"available", false},
{"renewing", false},
{"redundancy", 0},
{"uploadprogress", 0},
{"expiration", 0}
}));
uploadFileList->push_back(ptr);
}
else
{
uploadFileList->push_back(*it);
}
}
std::sort(uploadFileList->begin(), uploadFileList->end(), [](const CSiaFilePtr& a, const CSiaFilePtr& b)
{
return a->GetUploadProgress() > b->GetUploadProgress();
});
_uploadFileList = uploadFileList;
}
catch (const SQLite::Exception& e)
{
// error condition
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(DatabaseExceptionOccurred("AutoThreadCallback(uploadStatus)", e)));
}
// Lock here - if file is modified again before previously queued upload is complete, delete it and
// start again later
bool complete = false;
SString siaPath;
SString filePath;
std::lock_guard<std::mutex> l(_uploadMutex);
{
SQLite::Statement query(_uploadDatabase, QUERY_UPLOADS_BY_STATUS);
query.bind("@status", static_cast<unsigned>(UploadStatus::Uploading));
if (query.executeStep())
{
siaPath = static_cast<const char*>(query.getColumn(query.getColumnIndex("sia_path")));
filePath = static_cast<const char*>(query.getColumn(query.getColumnIndex("file_path")));
auto fileList = fileTree->GetFileList();
auto it = std::find_if(fileList->begin(), fileList->end(), [&](const CSiaFilePtr& ptr)
{
return ptr->GetSiaPath() == siaPath;
});
// Removed by another client
if (it == fileList->end())
{
HandleFileRemove(siaCurl, siaPath);
}
// Upload is complete
else if ((*it)->GetAvailable())
{
complete = true;
}
// Upload still active, don't process another file
else
{
SQLite::Statement count(_uploadDatabase, QUERY_UPLOAD_COUNT_BY_STATUS);
count.bind("@status", static_cast<unsigned>(UploadStatus::Uploading));
processNext = (count.executeStep() && (count.getColumn(0).getInt64() < _siaDriveConfig->GetMaxUploadCount()));
}
}
}
if (complete)
{
SET_STATUS(UploadStatus::Complete, UploadToSiaComplete, ModifyUploadStatusFailed)
processNext = statusUpdated;
}
}
else
{
// error condition - host down?
processNext = false;
}
}
catch (const SQLite::Exception& e)
{
// error condition - database not initialized (i.e. no table)?
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(DatabaseExceptionOccurred("AutoThreadCallback(/renter/files)", e)));
processNext = false;
}
if (processNext)
{
try
{
SString siaPath;
SString filePath;
bool uploading = false;
std::lock_guard<std::mutex> l(_uploadMutex);
{
SQLite::Statement query(_uploadDatabase, QUERY_UPLOADS_BY_STATUS);
query.bind("@status", static_cast<unsigned>(UploadStatus::Queued));
if (query.executeStep())
{
siaPath = static_cast<const char*>(query.getColumn(query.getColumnIndex("sia_path")));
filePath = static_cast<const char*>(query.getColumn(query.getColumnIndex("file_path")));
if (FilePath(filePath).IsFile())
{
json response;
SiaCurlError cerror = siaCurl.Post(SString(L"/renter/upload/") + siaPath, { {L"source", filePath} }, response);
if (ApiSuccess(cerror) || cerror.GetReason().Contains("already exists"))
{
uploading = !cerror.GetReason().Contains("already exists");
}
}
else
{
HandleFileRemove(siaCurl, siaPath);
}
}
else
{
processNext = false;
}
}
if (uploading)
{
SET_STATUS(UploadStatus::Uploading, UploadToSiaStarted, ModifyUploadStatusFailed)
processNext = statusUpdated;
}
}
catch (const SQLite::Exception& e)
{
// error condition
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(DatabaseExceptionOccurred("AutoThreadCallback(processNext)", e)));
processNext = false;
}
}
}
while (processNext);
}
UploadStatus CUploadManager::GetUploadStatus(const SString& siaPath)
{
UploadStatus uploadStatus = UploadStatus::NotFound;
SQLite::Statement query(_uploadDatabase, QUERY_UPLOADS_BY_SIA_PATH);
query.bind("@sia_path", SString::ToUtf8(siaPath).c_str());
if (query.executeStep())
{
uploadStatus = static_cast<UploadStatus>(static_cast<unsigned>(query.getColumn(query.getColumnIndex("status"))));
}
return uploadStatus;
}
UploadError CUploadManager::AddOrUpdate(const SString& siaPath, SString filePath, const std::uint64_t& lastModified)
{
UploadError ret;
// Relative to absolute and grab parent folder of source
FilePath rootPath = filePath;
rootPath.MakeAbsolute().RemoveFileName();
if (FilePath(filePath).IsFile())
{
// Lock here - if file is modified again before a prior upload is complete, delete it and
// start again later
std::lock_guard<std::mutex> l(_uploadMutex);
try
{
bool addToDatabase = true;
// Check upload status
{
SQLite::Statement query(_uploadDatabase, QUERY_UPLOADS_BY_SIA_PATH);
query.bind("@sia_path", SString::ToUtf8(siaPath).c_str());
if (query.executeStep())
{
std::uint64_t modifiedTime = query.getColumn(query.getColumnIndex("modified_time")).getInt64();
UploadStatus uploadStatus = static_cast<UploadStatus>(static_cast<unsigned>(query.getColumn(query.getColumnIndex("status"))));
addToDatabase = (uploadStatus == UploadStatus::Uploading);
if (!addToDatabase)
{
addToDatabase = (uploadStatus == UploadStatus::Complete) && (lastModified != modifiedTime);
if (addToDatabase)
{
HandleFileRemove(CSiaCurl(GetHostConfig()), siaPath);
}
}
}
else // Not in database, so assume is in Sia
{
HandleFileRemove(CSiaCurl(GetHostConfig()), siaPath);
}
}
if (addToDatabase)
{
try
{
// Add to db
SQLite::Statement insert(_uploadDatabase, INSERT_UPLOAD);
insert.bind("@sia_path", SString::ToUtf8(siaPath).c_str());
insert.bind("@file_path", SString::ToUtf8(filePath).c_str());
insert.bind("@modified_time", static_cast<long long>(lastModified));
insert.bind("@status", static_cast<unsigned>(UploadStatus::Queued));
if (insert.exec() == 1)
{
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(FileAddedToQueue(siaPath, filePath)));
}
else
{
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(DatabaseInsertFailed(siaPath, filePath, insert.getErrorMsg())));
ret = UploadErrorCode::DatabaseError;
}
}
catch (SQLite::Exception e)
{
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(DatabaseInsertFailed(siaPath, filePath, e.getErrorStr())));
ret = { UploadErrorCode::DatabaseError, e.getErrorStr() };
}
}
}
catch (SQLite::Exception e)
{
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(DatabaseExceptionOccurred("AddOrUpdate(query)", e)));
ret = { UploadErrorCode::DatabaseError, e.getErrorStr() };
}
}
else
{
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(SourceFileNotFound(siaPath, filePath)));
ret = UploadErrorCode::SourceFileNotFound;
}
return ret;
}
UploadError CUploadManager::Remove(const SString& siaPath)
{
UploadError ret;
try
{
std::lock_guard<std::mutex> l(_uploadMutex);
HandleFileRemove(CSiaCurl(GetHostConfig()), siaPath);
}
catch (SQLite::Exception e)
{
CEventSystem::EventSystem.NotifyEvent(CreateSystemEvent(DatabaseExceptionOccurred("Remove", e)));
ret = { UploadErrorCode::DatabaseError, e.getErrorStr() };
}
return ret;
}
UploadFileListPtr CUploadManager::GetUploadFileList() const
{
return _uploadFileList;
}