[Offline pages] Adding persistent request queue based on SQLite

components/offline_pages/background/request_queue_store_sql.cc

+// Copyright 2016 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "components/offline_pages/background/request_queue_store_sql.h"
+
+#include "base/bind.h"
+#include "base/files/file_path.h"
+#include "base/files/file_util.h"
+#include "base/location.h"
+#include "base/logging.h"
+#include "base/sequenced_task_runner.h"
+#include "base/threading/thread_task_runner_handle.h"
+#include "components/offline_pages/background/save_page_request.h"
+#include "sql/connection.h"
+#include "sql/statement.h"
+#include "sql/transaction.h"
+
+namespace offline_pages {
+
+namespace {
+
+// This is a macro instead of a const so that
+// it can be used inline in other SQL statements below.
+#define REQUEST_QUEUE_TABLE_NAME "request_queue_v1"
+
+// New columns should be added at the end of the list in order to avoid
+// complicated table upgrade.
+const char kOfflinePagesColumns[] =
+    "(request_id INTEGER PRIMARY KEY NOT NULL,"
+    " creation_time INTEGER NOT NULL,"
+    " activation_time INTEGER NOT NULL DEFAULT 0,"
+    " last_attempt_time INTEGER NOT NULL DEFAULT 0,"
+    " attempt_count INTEGER NOT NULL,"
+    " url VARCHAR NOT NULL,"
+    " client_namespace VARCHAR NOT NULL,"
+    " client_id VARCHAR NOT NULL"
+    ")";
+
+// Defines indices of the columns in a SELECT * FROM query. Should be kept in
+// sync with above.
+enum : int {
+  RQ_REQUEST_ID,
+  RQ_CREATION_TIME,
+  RQ_ACTIVATION_TIME,
+  RQ_LAST_ATTMEPT_TIME,
+  RQ_ATTEMPT_COUNT,
+  RQ_URL,
+  RQ_CLIENT_NAMESPACE,
+  RQ_CLIENT_ID,
+};
+
+enum class RequestExistsResult {
+  SQL_FAILED,
+  REQUEST_EXISTS,
+  REQUEST_DOES_NOT_EXIST,
+};
+
+// This is cloned from //content/browser/appcache/appcache_database.cc
+struct TableInfo {
+  const char* table_name;
+  const char* columns;
+};
+
+const TableInfo kRequestQueueTable{REQUEST_QUEUE_TABLE_NAME,
+                                   kOfflinePagesColumns};
+
+bool CreateTable(sql::Connection* db, const TableInfo& table_info) {
+  std::string sql("CREATE TABLE ");
+  sql += table_info.table_name;
+  sql += table_info.columns;
+  return db->Execute(sql.c_str());
+}
+
+bool CreateSchema(sql::Connection* db) {
+  // If you create a transaction but don't Commit() it is automatically
+  // rolled back by its destructor when it falls out of scope.
+  sql::Transaction transaction(db);
+  if (!transaction.Begin())
+    return false;
+
+  if (!db->DoesTableExist(kRequestQueueTable.table_name)) {
+    if (!CreateTable(db, kRequestQueueTable))
+      return false;
+  }
+
+  // TODO(fgorski): Add indices here.
+  return transaction.Commit();
+}
+
+bool DeleteByRequestId(sql::Connection* db, int64_t request_id) {
+  static const char kSql[] =
+      "DELETE FROM " REQUEST_QUEUE_TABLE_NAME " WHERE request_id=?";
+  sql::Statement statement(db->GetCachedStatement(SQL_FROM_HERE, kSql));
+  statement.BindInt64(0, request_id);
+  LOG(ERROR) << "In DeleteRequestById: "
+             << request_id;  // statement.GetSQLStatement();
+  return statement.Run();
+}
+
+// Create a save page request from a SQL result.  Expects complete rows with
+// all columns present.
+SavePageRequest MakeSavePageRequest(sql::Statement* statement) {
+  int64_t id = statement->ColumnInt64(RQ_REQUEST_ID);
+  GURL url(statement->ColumnString(RQ_URL));
+  ClientId client_id(statement->ColumnString(RQ_CLIENT_NAMESPACE),
+                     statement->ColumnString(RQ_CLIENT_ID));
+  int64_t attempt_count = statement->ColumnInt64(RQ_ATTEMPT_COUNT);
+  base::Time creation_time =
+      base::Time::FromInternalValue(statement->ColumnInt64(RQ_CREATION_TIME));
+  base::Time activation_time =
+      base::Time::FromInternalValue(statement->ColumnInt64(RQ_ACTIVATION_TIME));
+
+  SavePageRequest request(id, url, client_id, creation_time, activation_time);
+  base::Time last_attempt_time = base::Time::FromInternalValue(
+      statement->ColumnInt64(RQ_LAST_ATTMEPT_TIME));
+  request.set_last_attempt_time(last_attempt_time);
+  request.set_attempt_count(attempt_count);
+
+  return request;
+}
+
+RequestExistsResult RequestExists(sql::Connection* db, int64_t request_id) {
+  const char kSql[] =
+      "SELECT COUNT(*) FROM " REQUEST_QUEUE_TABLE_NAME " WHERE request_id = ?";
+  sql::Statement statement(db->GetUniqueStatement(kSql));
+  statement.BindInt64(0, request_id);
+  if (!statement.Step()) {
+    LOG(ERROR) << "Failed to check if request exists: " << request_id
+               << ", SQL statement: " << statement.GetSQLStatement();
+    return RequestExistsResult::SQL_FAILED;
+  }
+  return statement.ColumnInt64(0) ? RequestExistsResult::REQUEST_EXISTS
+                                  : RequestExistsResult::REQUEST_DOES_NOT_EXIST;
+}
+
+RequestQueueStore::UpdateStatus InsertOrReplace(
+    sql::Connection* db,
+    const SavePageRequest& request) {
+  // In order to use the enums in the Bind* methods, keep the order of fields
+  // the same as in the definition/select query.
+  const char kInsertSql[] =
+      "INSERT OR REPLACE INTO " REQUEST_QUEUE_TABLE_NAME
+      " (request_id, creation_time, activation_time, last_attempt_time, "
+      " attempt_count, url, client_namespace, client_id) "
+      " VALUES "
+      " (?, ?, ?, ?, ?, ?, ?, ?)";
+
+  // Checking if a request exists and adding/updating it will happen in a single
+  // transaction.
+  sql::Transaction transaction(db);
+  if (!transaction.Begin()) {
+    LOG(ERROR) << "Failed to start transaction on InsertOrReplace.";
+    return RequestQueueStore::UpdateStatus::FAILED;
+  }
+
+  RequestExistsResult exists = RequestExists(db, request.request_id());
+  if (exists == RequestExistsResult::SQL_FAILED) {
+    LOG(ERROR) << "Failed to check if request exists: " << request.request_id();
+    return RequestQueueStore::UpdateStatus::FAILED;
+  }
+
+  RequestQueueStore::UpdateStatus status =
+      exists == RequestExistsResult::REQUEST_EXISTS
+          ? RequestQueueStore::UpdateStatus::UPDATED
+          : RequestQueueStore::UpdateStatus::ADDED;
+
+  sql::Statement statement(db->GetCachedStatement(SQL_FROM_HERE, kInsertSql));
+  statement.BindInt64(RQ_REQUEST_ID, request.request_id());
+  statement.BindString(RQ_URL, request.url().spec());
+  statement.BindString(RQ_CLIENT_NAMESPACE, request.client_id().name_space);
+  statement.BindString(RQ_CLIENT_ID, request.client_id().id);
+  statement.BindInt64(RQ_CREATION_TIME,
+                      request.creation_time().ToInternalValue());
+  statement.BindInt64(RQ_ACTIVATION_TIME,
+                      request.activation_time().ToInternalValue());
+  statement.BindInt64(RQ_LAST_ATTMEPT_TIME,
+                      request.last_attempt_time().ToInternalValue());
+  statement.BindInt64(RQ_ATTEMPT_COUNT, request.attempt_count());
+
+  if (!statement.Run() || !transaction.Commit()) {
+    LOG(ERROR) << "Failed to add/update a request: " << request.request_id();
+    return RequestQueueStore::UpdateStatus::FAILED;
+  }
+
+  return status;
+}
+
+bool InitDatabase(sql::Connection* db, const base::FilePath& path) {
+  db->set_page_size(4096);
+  db->set_cache_size(500);
+  db->set_histogram_tag("BackgroundRequestQueue");
+  db->set_exclusive_locking();
+
+  base::File::Error err;
+  if (!base::CreateDirectoryAndGetError(path.DirName(), &err)) {
+    LOG(ERROR) << "Failed to create background request queue db directory: "
+               << base::File::ErrorToString(err);
+    return false;
+  }
+  if (!db->Open(path)) {
+    LOG(ERROR) << "Failed to open database";
+    return false;
+  }
+  db->Preload();
+
+  return CreateSchema(db);
+}
+
+}  // anonymous namespace
+
+RequestQueueStoreSQL::RequestQueueStoreSQL(
+    scoped_refptr<base::SequencedTaskRunner> background_task_runner,
+    const base::FilePath& path)
+    : background_task_runner_(std::move(background_task_runner)),
+      db_file_path_(path.AppendASCII("RequestQueue.db")),
+      weak_ptr_factory_(this) {
+  OpenConnection();
+}
+
+RequestQueueStoreSQL::~RequestQueueStoreSQL() {
+  if (db_.get() &&
+      !background_task_runner_->DeleteSoon(FROM_HERE, db_.release())) {
+    DLOG(WARNING) << "SQL database will not be deleted.";
+  }
+}
+
+// static
+void RequestQueueStoreSQL::OpenConnectionSync(
+    sql::Connection* db,
+    scoped_refptr<base::SingleThreadTaskRunner> runner,
+    const base::FilePath& path,
+    const base::Callback<void(bool)>& callback) {
+  bool success = InitDatabase(db, path);
+  runner->PostTask(FROM_HERE, base::Bind(callback, success));
+}
+
+// static
+void RequestQueueStoreSQL::GetRequestsSync(
+    sql::Connection* db,
+    scoped_refptr<base::SingleThreadTaskRunner> runner,
+    const GetRequestsCallback& callback) {
+  const char kSql[] = "SELECT * FROM " REQUEST_QUEUE_TABLE_NAME;
+
+  sql::Statement statement(db->GetCachedStatement(SQL_FROM_HERE, kSql));
+
+  std::vector<SavePageRequest> result;
+  while (statement.Step()) {
+    result.push_back(MakeSavePageRequest(&statement));
+  }
+
+  runner->PostTask(FROM_HERE,
+                   base::Bind(callback, statement.Succeeded(), result));
+}
+
+// static
+void RequestQueueStoreSQL::AddOrUpdateRequestSync(
+    sql::Connection* db,
+    scoped_refptr<base::SingleThreadTaskRunner> runner,
+    const SavePageRequest& request,
+    const UpdateCallback& callback) {
+  // TODO(fgorski): add UMA metrics here.
+  RequestQueueStore::UpdateStatus status = InsertOrReplace(db, request);
+  runner->PostTask(FROM_HERE, base::Bind(callback, status));
+}
+
+// static
+void RequestQueueStoreSQL::RemoveRequestsSync(
+    sql::Connection* db,
+    scoped_refptr<base::SingleThreadTaskRunner> runner,
+    const std::vector<int64_t>& request_ids,
+    const RemoveCallback& callback) {
+  // TODO(fgorski): add UMA metrics here.
+
+  // If you create a transaction but don't Commit() it is automatically
+  // rolled back by its destructor when it falls out of scope.
+  sql::Transaction transaction(db);
+  if (!transaction.Begin()) {
+    runner->PostTask(FROM_HERE, base::Bind(callback, false, 0));
+    return;
+  }
+  int count = 0;
+  for (auto request_id : request_ids) {
+    RequestExistsResult exists = RequestExists(db, request_id);
+    if (exists == RequestExistsResult::SQL_FAILED) {
+      runner->PostTask(FROM_HERE, base::Bind(callback, false, 0));
+      return;
+    }
+    if (exists == RequestExistsResult::REQUEST_DOES_NOT_EXIST)
+      continue;
+
+    ++count;
+    if (!DeleteByRequestId(db, request_id)) {
+      runner->PostTask(FROM_HERE, base::Bind(callback, false, 0));
+      return;
+    }
+  }
+
+  if (!transaction.Commit()) {
+    runner->PostTask(FROM_HERE, base::Bind(callback, false, 0));
+    return;
+  }
+
+  runner->PostTask(FROM_HERE, base::Bind(callback, true, count));
+}
+
+// static
+void RequestQueueStoreSQL::ResetSync(
+    std::unique_ptr<sql::Connection> db,
+    scoped_refptr<base::SingleThreadTaskRunner> runner,
+    const ResetCallback& callback) {
+  // This method deletes the content of the whole store. It should be used with
+  // caution, e.g. when recovering from situation, in which the contents of the
+  // store does not make sense and cannot be converted to reasonable requests.
+  const char kSql[] = "DELETE FROM " REQUEST_QUEUE_TABLE_NAME;
+  sql::Statement statement(db->GetCachedStatement(SQL_FROM_HERE, kSql));
+  runner->PostTask(FROM_HERE, base::Bind(callback, statement.Run()));
+}
+
+void RequestQueueStoreSQL::GetRequests(const GetRequestsCallback& callback) {
+  DCHECK(db_.get());
+  if (!db_.get()) {
+    // Nothing to do, but post a callback instead of calling directly
+    // to preserve the async style behavior to prevent bugs.
+    base::ThreadTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE, base::Bind(callback, false, std::vector<SavePageRequest>()));
+    return;
+  }
+
+  background_task_runner_->PostTask(
+      FROM_HERE, base::Bind(&RequestQueueStoreSQL::GetRequestsSync, db_.get(),
+                            base::ThreadTaskRunnerHandle::Get(), callback));
+}
+
+void RequestQueueStoreSQL::AddOrUpdateRequest(const SavePageRequest& request,
+                                              const UpdateCallback& callback) {
+  DCHECK(db_.get());
+  if (!db_.get()) {
+    // Nothing to do, but post a callback instead of calling directly
+    // to preserve the async style behavior to prevent bugs.
+    base::ThreadTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE, base::Bind(callback, UpdateStatus::FAILED));
+    return;
+  }
+
+  background_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&RequestQueueStoreSQL::AddOrUpdateRequestSync, db_.get(),
+                 base::ThreadTaskRunnerHandle::Get(), request, callback));
+}
+
+void RequestQueueStoreSQL::RemoveRequests(
+    const std::vector<int64_t>& request_ids,
+    const RemoveCallback& callback) {
+  DCHECK(db_.get());
+  if (!db_.get()) {
+    // Nothing to do, but post a callback instead of calling directly
+    // to preserve the async style behavior to prevent bugs.
+    base::ThreadTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE, base::Bind(callback, false, 0));
+    return;
+  }
+
+  if (request_ids.empty()) {
+    // Nothing to do, but post a callback instead of calling directly
+    // to preserve the async style behavior to prevent bugs.
+    base::ThreadTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE, base::Bind(callback, true, 0));
+    return;
+  }
+
+  background_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&RequestQueueStoreSQL::RemoveRequestsSync, db_.get(),
+                 base::ThreadTaskRunnerHandle::Get(), request_ids, callback));
+}
+
+void RequestQueueStoreSQL::Reset(const ResetCallback& callback) {
+  DCHECK(db_.get());
+  if (!db_.get()) {
+    // Nothing to do, but post a callback instead of calling directly
+    // to preserve the async style behavior to prevent bugs.
+    base::ThreadTaskRunnerHandle::Get()->PostTask(FROM_HERE,
+                                                  base::Bind(callback, false));
+    return;
+  }
+
+  background_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&RequestQueueStoreSQL::ResetSync, base::Passed(&db_),
+                 base::ThreadTaskRunnerHandle::Get(), callback));
+}
+
+void RequestQueueStoreSQL::OpenConnection() {
+  DCHECK(!db_);
+  db_.reset(new sql::Connection());
+  background_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&RequestQueueStoreSQL::OpenConnectionSync, db_.get(),
+                 base::ThreadTaskRunnerHandle::Get(), db_file_path_,
+                 base::Bind(&RequestQueueStoreSQL::OnOpenConnectionDone,
+                            weak_ptr_factory_.GetWeakPtr())));
+}
+
+void RequestQueueStoreSQL::OnOpenConnectionDone(bool success) {
+  DCHECK(db_.get());
+
+  // Unfortunately we were not able to open DB connection.
+  if (!success) {
+    LOG(ERROR) << "Database creation fialed.";
+    db_.reset();
+  }
+}
+
+}  // namespace offline_pages