int
block_tx_client_start (BlockTxInfo *info, BlockTxClientDoneCB cb)
{
    BlockTxClient *client = g_new0 (BlockTxClient, 1);
    int ret = 0;

    client->info = info;
    client->cb = cb;

    ret = ccnet_job_manager_schedule_job (seaf->job_mgr,
                                          block_tx_client_thread,
                                          block_tx_client_thread_done,
                                          client);
    if (ret < 0) {
        seaf_warning ("Failed to start block tx client thread.\n");
        return -1;
    }

    return 0;
}

static void command_read_cb (CFFileDescriptorRef fdref,
                             CFOptionFlags callBackTypes,
                             void *info)
{
    SeafWTMonitorPriv *priv = (SeafWTMonitorPriv *)info;
    WatchCommand cmd;
    int n;

    n = pipereadn (priv->cmd_pipe[0], &cmd, sizeof(cmd));
    if (n != sizeof(cmd)) {
        seaf_warning ("[wt mon] failed to read command.\n");
        CFFileDescriptorEnableCallBacks (fdref, kCFFileDescriptorReadCallBack);
        return;
    }

    seaf_debug ("[wt mon] %ld receive command type=%d, repo=%s\n",
                (long)CFRunLoopGetCurrent(), cmd.type, cmd.repo_id);
    handle_watch_command (priv, &cmd);
    CFFileDescriptorEnableCallBacks (fdref, kCFFileDescriptorReadCallBack);
}

static void
on_fs_write (OSAsyncResult *res, void *cb_data)
{
    CcnetProcessor *processor = cb_data;
    USE_PRIV;

    if (!res->success) {
        seaf_warning ("[recvfs] Failed to write %s.\n", res->obj_id);
        ccnet_processor_send_response (processor, SC_BAD_OBJECT, SS_BAD_OBJECT,
                                       NULL, 0);
        ccnet_processor_done (processor, FALSE);
        return;
    }

    --priv->pending_objects;

#ifdef DEBUG
    seaf_debug ("[recvfs] Wrote fs object %s.\n", res->obj_id);
#endif
}

GList *
seaf_share_manager_list_shared_to (SeafShareManager *mgr,
                                   const char *owner,
                                   const char *repo_id)
{
    char *sql;
    GList *ret = NULL;

    sql = "SELECT to_email FROM SharedRepo WHERE "
        "from_email=? AND repo_id=?";
    if (seaf_db_statement_foreach_row (mgr->seaf->db, sql,
                                       collect_shared_to, &ret,
                                       2, "string", owner, "string", repo_id) < 0) {
        seaf_warning ("[share mgr] DB error when list shared to.\n");
        string_list_free (ret);
        return NULL;
    }

    return ret;
}

static gboolean
case_conflict_exists (const char *dir_path, const char *new_dname,
                      char **conflict_dname)
{
    GDir *dir;
    const char *dname;
    gboolean is_case_conflict = FALSE;
    GError *error = NULL;

    dir = g_dir_open (dir_path, 0, &error);
    if (!dir && error) {
        seaf_warning ("Failed to open dir %s: %s.\n", dir_path, error->message);
        g_error_free (error);
        return FALSE;
    }

    while (1) {
        dname = g_dir_read_name (dir);
        if (!dname)
            break;

#ifdef __APPLE__
        char *norm_dname = g_utf8_normalize (dname, -1, G_NORMALIZE_NFC);
        if (case_conflict_utf8 (norm_dname, new_dname)) {
            is_case_conflict = TRUE;
            *conflict_dname = norm_dname;
            break;
        }
        g_free (norm_dname);
#else
        if (case_conflict_utf8 (dname, new_dname)) {
            is_case_conflict = TRUE;
            *conflict_dname = g_strdup(dname);
            break;
        }
#endif
    }
    g_dir_close (dir);

    return is_case_conflict;
}

static int
start (CcnetProcessor *processor, int argc, char **argv)
{
    USE_PRIV;
    GString *buf;
    SeafileGetfsV2Proc *proc = (SeafileGetfsV2Proc *)processor;
    TransferTask *task = proc->tx_task;

    buf = g_string_new (NULL);
    if (!task->is_clone) {
        SeafBranch *master = seaf_branch_manager_get_branch (seaf->branch_mgr,
                                                             task->repo_id,
                                                             "master");
        if (!master) {
            seaf_warning ("Master branch not found for repo %s.\n", task->repo_id);
            g_string_free (buf, TRUE);
            ccnet_processor_done (processor, FALSE);
            return -1;
        }

        g_string_printf (buf, "remote %s seafile-putfs-v2 %s %s %s",
                         processor->peer_id, task->session_token,
                         task->head, master->commit_id);

        seaf_branch_unref (master);
    } else
        g_string_printf (buf, "remote %s seafile-putfs-v2 %s %s",
                         processor->peer_id, task->session_token,
                         task->head);

    ccnet_processor_send_request (processor, buf->str);
    g_string_free (buf, TRUE);

    priv->registered = TRUE;
    priv->writer_id = seaf_obj_store_register_async_write (seaf->fs_mgr->obj_store,
                                                           task->repo_id,
                                                           task->repo_version,
                                                           on_fs_write,
                                                           processor);
    return 0;
}

static void handle_response (CcnetProcessor *processor,
                             char *code, char *code_msg,
                             char *content, int clen)
{
    SeafileGetcommitV3Proc *proc = (SeafileGetcommitV3Proc *)processor;

    if (proc->tx_task->state != TASK_STATE_NORMAL) {
        ccnet_processor_done (processor, TRUE);
        return;
    }

    if (strncmp(code, SC_OK, 3) == 0) {
        receive_commit (processor, content, clen);
        return;
    }

    seaf_warning ("Bad response: %s %s.\n", code, code_msg);
    if (memcmp (code, SC_ACCESS_DENIED, 3) == 0)
        transfer_task_set_error (proc->tx_task, TASK_ERR_ACCESS_DENIED);
    ccnet_processor_done (processor, FALSE);
}

static int
check_fs_tree_from (CcnetProcessor *processor, const char *root_id)
{
    USE_PRIV;

    memcpy (priv->root_id, root_id, 40);
    priv->fetch_objs = NULL;

    int rc = ccnet_processor_thread_create (processor,
                                            seaf->job_mgr,
                                            check_objects_thread,
                                            check_objects_done,
                                            processor);
    if (rc < 0) {
        seaf_warning ("Failed to start worker thread.\n");
        return -1;
    }

    priv->worker_running = TRUE;
    return 0;
}

static gint64
repo_share_usage (const char *user, const char *repo_id)
{
    gint n_shared_to = get_num_shared_to (user, repo_id);
    if (n_shared_to < 0) {
        return -1;
    } else if (n_shared_to == 0) {
        return 0;
    }

    gint64 size = seaf_repo_manager_get_repo_size (seaf->repo_mgr, repo_id);
    if (size < 0) {
        seaf_warning ("Cannot get size of repo %s.\n", repo_id);
        return -1;
    }

    /* share_usage = repo_size * n_shared_to */
    gint64 usage = size * n_shared_to;

    return usage;
}

static void
load_repo_commit (SeafRepoManager *manager,
                  SeafRepo *repo,
                  SeafBranch *branch)
{
    SeafCommit *commit;

    commit = seaf_commit_manager_get_commit_compatible (manager->seaf->commit_mgr,
                                                        repo->id,
                                                        branch->commit_id);
    if (!commit) {
        seaf_warning ("Commit %s is missing\n", branch->commit_id);
        repo->is_corrupted = TRUE;
        return;
    }

    set_head_common (repo, branch);
    seaf_repo_from_commit (repo, commit);

    seaf_commit_unref (commit);
}

static BMetadata *
block_backend_fs_stat_block (BlockBackend *bend,
                             const char *store_id,
                             int version,
                             const char *block_id)
{
    char path[SEAF_PATH_MAX];
    SeafStat st;
    BMetadata *block_md;

    get_block_path (bend, block_id, path, store_id, version);
    if (seaf_stat (path, &st) < 0) {
        seaf_warning ("[block bend] Failed to stat block %s.\n", block_id);
        return NULL;
    }
    block_md = g_new0(BMetadata, 1);
    memcpy (block_md->id, block_id, 40);
    block_md->size = (uint32_t) st.st_size;

    return block_md;
}

static void handle_update (CcnetProcessor *processor,
                           char *code, char *code_msg,
                           char *content, int clen)
{
    switch (processor->state) {
    case PREPARE:
        if (memcmp (code, SC_BLOCKLIST, 3) == 0) {
            process_block_list (processor, content, clen);
            return;
        } else if (memcmp (code, SC_GET_PORT, 3) == 0) {
            send_port (processor);
            return;
        }
        break;
    }

    seaf_warning ("Bad code: %s %s\n", code, code_msg);
    ccnet_processor_send_response (processor, SC_BAD_UPDATE_CODE, 
                                   SS_BAD_UPDATE_CODE, NULL, 0);
    ccnet_processor_done (processor, FALSE);
}

static void
init_seafile_path ()
{
    GError *error = NULL;
    char *binary = g_file_read_link ("/proc/self/exe", &error);
    char *tmp = NULL;
    if (error != NULL) {
        seaf_warning ("failed to readlink: %s\n", error->message);
        return;
    }

    bin_dir = g_path_get_dirname (binary);

    tmp = g_path_get_dirname (bin_dir);
    installpath = g_path_get_dirname (tmp);

    topdir = g_path_get_dirname (installpath);

    g_free (binary);
    g_free (tmp);
}

static void
start_upload_if_necessary (SyncTask *task)
{
    GError *error = NULL;
    const char *repo_id = task->repo->id;

    char *tx_id = seaf_transfer_manager_add_upload (seaf->transfer_mgr,
                                                    repo_id,
                                                    task->dest_id,
                                                    "local",
                                                    "master",
                                                    task->token,
                                                    &error);
    if (error != NULL) {
        seaf_warning ("Failed to start upload: %s\n", error->message);
        seaf_sync_manager_set_task_error (task, SYNC_ERROR_START_UPLOAD);
        return;
    }
    task->tx_id = tx_id;
    transition_sync_state (task, SYNC_STATE_UPLOAD);
}

static void *
get_repo_info_thread (void *data)
{
    CcnetProcessor *processor = data;
    USE_PRIV;
    SeafRepo *repo;

    repo = seaf_repo_manager_get_repo (seaf->repo_mgr, priv->repo_id);
    if (!repo) {
        seaf_warning ("Failed to get repo %s.\n", priv->repo_id);
        priv->success = FALSE;
        return data;
    }

    memcpy (priv->store_id, repo->store_id, 36);
    priv->repo_version = repo->version;
    priv->success = TRUE;

    seaf_repo_unref (repo);
    return data;
}

int
seafile_update_random_key (const char *old_passwd, const char *old_random_key,
                           const char *new_passwd, char *new_random_key)
{
    unsigned char key[32], iv[16];
    unsigned char random_key_raw[48], *secret_key, *new_random_key_raw;
    int secret_key_len, random_key_len;
    SeafileCrypt *crypt;

    /* First, use old_passwd to decrypt secret key from old_random_key. */
    seafile_derive_key (old_passwd, strlen(old_passwd), 2, key, iv);

    hex_to_rawdata (old_random_key, random_key_raw, 48);

    crypt = seafile_crypt_new (2, key, iv);
    if (seafile_decrypt ((char **)&secret_key, &secret_key_len,
                         (char *)random_key_raw, 48,
                         crypt) < 0) {
        seaf_warning ("Failed to decrypt random key.\n");
        g_free (crypt);
        return -1;
    }
    g_free (crypt);

    /* Second, use new_passwd to encrypt secret key. */
    seafile_derive_key (new_passwd, strlen(new_passwd), 2, key, iv);

    crypt = seafile_crypt_new (2, key, iv);

    seafile_encrypt ((char **)&new_random_key_raw, &random_key_len,
                     (char *)secret_key, secret_key_len, crypt);

    rawdata_to_hex (new_random_key_raw, new_random_key, 48);

    g_free (secret_key);
    g_free (new_random_key_raw);
    g_free (crypt);

    return 0;
}

/* Add a specific HANDLE to an I/O Completion Port. If it's the cmd pipe
 * handle, call ReadFile() on it; If it's a dir handle, call
 * ReadDirectoryChangesW() on it.
 */
static BOOL
add_handle_to_iocp (SeafWTMonitor *monitor, HANDLE hAdd)
{
    SeafWTMonitorPriv *priv = monitor->priv;
    
    if (!priv || !hAdd)
        return FALSE;

    /* CreateIoCompletionPort() will add the handle to an I/O Completion Port
      if the iocp handle is not NULL. Otherwise it will create a new IOCP
      handle.

      The `key' parameter is used by th IOCP to tell us which handle watched
      by the I/O Completion Port triggeed a return of the
      GetQueuedCompletionStatus() function.

      Here we use the value of the handle itself as the key for this handle
      in the I/O Completion Port.
    */
    priv->iocp_handle = CreateIoCompletionPort
        (hAdd,                  /* handle to add */
         priv->iocp_handle,     /* iocp handle */
         (ULONG_PTR)hAdd,       /* key for this handle */
         1);                    /* Num of concurrent threads */

    if (!priv->iocp_handle) {
        seaf_warning ("failed to create/add iocp, error code %lu",
                      GetLastError());
        return FALSE;
    }

    if (hAdd == (HANDLE)monitor->cmd_pipe[0]) {
        /* HANDLE is cmd_pipe */
        return start_watch_cmd_pipe (monitor, NULL);
    } else {
        /* HANDLE is a dir handle */
        return start_watch_dir_change (priv, hAdd);
    }

}

static void
on_fs_write (OSAsyncResult *res, void *cb_data)
{
    CcnetProcessor *processor = cb_data;
    USE_PRIV;

    if (!res->success) {
        seaf_warning ("[recvfs] Failed to write %s.\n", res->obj_id);
        ccnet_processor_send_response (processor, SC_BAD_OBJECT, SS_BAD_OBJECT,
                                       NULL, 0);
        ccnet_processor_done (processor, FALSE);
        return;
    }

    seaf_debug ("[recvfs] Wrote fs object %s.\n", res->obj_id);

    if (++(priv->n_saved) == priv->total_needed) {
        seaf_debug ("All objects saved. Done.\n");
        ccnet_processor_send_response (processor, SC_END, SS_END, NULL, 0);
        ccnet_processor_done (processor, TRUE);
    }
}

static gboolean
fs_callback (SeafFSManager *mgr,
             const char *store_id,
             int version,
             const char *obj_id,
             int type,
             void *user_data,
             gboolean *stop)
{
    MigrationData *data = user_data;
    SeafRepo *repo = data->repo;

    if (data->visited != NULL) {
        if (g_hash_table_lookup (data->visited, obj_id) != NULL) {
            *stop = TRUE;
            return TRUE;
        }

        char *key = g_strdup(obj_id);
        g_hash_table_insert (data->visited, key, key);
    }

    if (seaf_obj_store_copy_obj (seaf->fs_mgr->obj_store,
                                 repo->store_id, repo->version,
                                 repo->store_id, 1,
                                 obj_id) < 0) {
        seaf_warning ("Failed to copy fs object %s.\n", obj_id);
        return FALSE;
    }

    if (data->stop_copy_blocks)
        return TRUE;

    if (type == SEAF_METADATA_TYPE_FILE &&
        migrate_file_blocks (mgr, data, obj_id) < 0)
        return FALSE;

    return TRUE;
}

GList *
seaf_repo_get_commits (SeafRepo *repo)
{
    GList *branches;
    GList *ptr;
    SeafBranch *branch;
    GList *commits = NULL;

    branches = seaf_branch_manager_get_branch_list (seaf->branch_mgr, repo->id);
    if (branches == NULL) {
        seaf_warning ("Failed to get branch list of repo %s.\n", repo->id);
        return NULL;
    }

    for (ptr = branches; ptr != NULL; ptr = ptr->next) {
        branch = ptr->data;
        gboolean res = seaf_commit_manager_traverse_commit_tree (seaf->commit_mgr,
                                                                 repo->id,
                                                                 repo->version,
                                                                 branch->commit_id,
                                                                 collect_commit,
                                                                 &commits,
                                                                 FALSE);
        if (!res) {
            for (ptr = commits; ptr != NULL; ptr = ptr->next)
                seaf_commit_unref ((SeafCommit *)(ptr->data));
            g_list_free (commits);
            goto out;
        }
    }

    commits = g_list_reverse (commits);

out:
    for (ptr = branches; ptr != NULL; ptr = ptr->next) {
        seaf_branch_unref ((SeafBranch *)ptr->data);
    }
    return commits;
}