/*
 * On Windows, RENAMED_OLD_NAME and RENAMED_NEW_NAME always comes in pairs.
 * If a file or dir is moved in/out of the worktree, ADDED or REMOVED event
 * will be emitted by the kernel.
 * 
 * This is a two-state state machine. The states are 'not processing rename' and
 * 'processing rename'.
 */
static void
handle_rename (RepoWatchInfo *info,
               PFILE_NOTIFY_INFORMATION event,
               const char *worktree,
               const char *filename,
               gboolean last_event)
{
    WTStatus *status = info->status;
    RenameInfo *rename_info = info->rename_info;

    if (event->Action == FILE_ACTION_RENAMED_OLD_NAME)
        seaf_debug ("Move %s ->\n", filename);
    else if (event->Action == FILE_ACTION_RENAMED_NEW_NAME)
        seaf_debug ("Move -> %s.\n", filename);

    if (!rename_info->processing) {
        if (event->Action == FILE_ACTION_RENAMED_OLD_NAME) {
            if (!last_event) {
                set_rename_processing_state (rename_info, filename);
            } else {
                /* RENAMED_OLD_NAME should not be the last event,
                   just ignore it.
                */
            }
        }
    } else {
        if (event->Action == FILE_ACTION_RENAMED_NEW_NAME) {
            /* Rename pair detected. */
            add_event_to_queue (status, WT_EVENT_RENAME,
                                rename_info->old_path, filename);
            unset_rename_processing_state (rename_info);
        }
    }
}

static evhtp_res
recv_form_field (RecvFSM *fsm, gboolean *no_line)
{
    char *line;
    size_t len;

    *no_line = FALSE;

    line = evbuffer_readln (fsm->line, &len, EVBUFFER_EOL_CRLF_STRICT);
    if (line != NULL) {
        if (strstr (line, fsm->boundary) != NULL) {
            seaf_debug ("[upload] form field ends.\n");

            g_free (fsm->input_name);
            fsm->input_name = NULL;
            fsm->state = RECV_HEADERS;
        } else {
            seaf_debug ("[upload] form field is %s.\n", line);

            g_hash_table_insert (fsm->form_kvs,
                                 g_strdup(fsm->input_name),
                                 g_strdup(line));
        }
        free (line);
    } else {
        *no_line = TRUE;
    }

    return EVHTP_RES_OK;
}

static void *
compute_upload_commits_thread (void *vdata)
{
    CcnetProcessor *processor = vdata;
    SeafileSendcommitV3Proc *proc = (SeafileSendcommitV3Proc *)processor;
    TransferTask *task = proc->tx_task;
    USE_PRIV;
    gboolean ret;

    priv->fast_forward = TRUE;
    ret = seaf_commit_manager_traverse_commit_tree (seaf->commit_mgr,
                                                    task->head,
                                                    traverse_commit_fast_forward,
                                                    processor, FALSE);
    if (!ret) {
        priv->compute_success = FALSE;
        return vdata;
    }

    if (priv->fast_forward) {
        priv->compute_success = TRUE;
        seaf_debug ("[sendcommt] Send commit after a fast forward merge.\n");
        return vdata;
    }

    seaf_debug ("[sendcommit] Send commit after a real merge.\n");
    if (compute_delta_commits (processor, task->head) < 0) {
        priv->compute_success = FALSE;
        return vdata;
    }

    priv->compute_success = TRUE;
    return vdata;
}

static void *
collect_commit_id_thread (void *vprocessor)
{
    CcnetProcessor *processor = vprocessor;
    USE_PRIV;

    priv->fast_forward = TRUE;
    if (seaf_commit_manager_traverse_commit_tree (seaf->commit_mgr,
                                                  priv->head_commit_id,
                                                  collect_id_fast_forward,
                                                  processor,
                                                  FALSE) < 0) {
        g_warning ("[putcommit] Failed to collect commit id.\n");
        string_list_free (priv->id_list);
        priv->id_list = NULL;
        return vprocessor;
    }

    if (priv->fast_forward) {
        seaf_debug ("Send commits after a fast-forward merge.\n");
        return vprocessor;
    }

    seaf_debug ("Send commits after a real merge.\n");

    compute_delta_commits (processor, priv->head_commit_id);

    return vprocessor;
}

static gboolean
send_fs_object (CcnetProcessor *processor, char *object_id)
{
    char *data;
    int len;
    ObjectPack *pack = NULL;
    int pack_size;

    if (seaf_obj_store_read_obj (seaf->fs_mgr->obj_store,
                                 object_id, (void**)&data, &len) < 0) {
        g_warning ("Failed to read fs object %s.\n", object_id);
        goto fail;
    }

    pack_size = sizeof(ObjectPack) + len;
    pack = malloc (pack_size);
    memcpy (pack->id, object_id, 41);
    memcpy (pack->object, data, len);

    if (pack_size <= MAX_OBJ_SEG_SIZE) {
        ccnet_processor_send_update (processor, SC_OBJECT, SS_OBJECT,
                                     (char *)pack, pack_size);
    } else {
        int offset, n;

        offset = 0;
        while (offset < pack_size) {
            n = MIN(pack_size - offset, MAX_OBJ_SEG_SIZE);

            if (offset + n < pack_size) {
                ccnet_processor_send_update (processor,
                                             SC_OBJ_SEG, SS_OBJ_SEG,
                                             (char *)pack + offset, n);
            } else {
                ccnet_processor_send_update (processor,
                                             SC_OBJ_SEG_END, SS_OBJ_SEG_END,
                                             (char *)pack + offset, n);
            }

            seaf_debug ("Sent object %s segment<total = %d, offset = %d, n = %d>\n",
                        object_id, pack_size, offset, n);

            offset += n;
        }
    }

    seaf_debug ("Send fs object %.8s.\n", object_id);

    g_free (data);
    free (pack);
    return TRUE;

fail:
    ccnet_processor_send_update (processor, SC_NOT_FOUND, SS_NOT_FOUND,
                                 object_id, 41);
    ccnet_processor_done (processor, FALSE);
    return FALSE;
}

static int
handle_handshake_response (BlockTxClient *client)
{
    BlockTxInfo *info = client->info;
    struct evbuffer *input = client->recv_buf;
    HandshakeResponse rsp;

    if (evbuffer_get_length (input) < sizeof(rsp))
        return 0;

    evbuffer_remove (input, &rsp, sizeof(rsp));

    rsp.status = ntohl(rsp.status);
    rsp.version = ntohl(rsp.version);

    if (rsp.status == STATUS_OK) {
        seaf_debug ("Handshake OK.\n");

        client->version = MIN (rsp.version, BLOCK_PROTOCOL_VERSION);

        if (client->version == 1)
            blocktx_generate_encrypt_key (info->session_key, sizeof(info->session_key),
                                          client->key, client->iv);
        else if (client->version == 2)
            blocktx_generate_encrypt_key (info->session_key, sizeof(info->session_key),
                                          client->key_v2, client->iv_v2);
        else {
            seaf_warning ("Bad block protocol version %d.\n", rsp.version);
            info->result = BLOCK_CLIENT_FAILED;
            return -1;
        }

        seaf_debug ("Block protocol version %d.\n", client->version);

        init_frame_parser (client);

        if (send_authentication (client) < 0)
            return -1;

        return 0;
    } else if (rsp.status == STATUS_VERSION_MISMATCH) {
        seaf_warning ("The server refuse to accpet protocol version %d.\n"
                      "Remote version is %d.\n",
                      BLOCK_PROTOCOL_VERSION, rsp.version);
        /* this is a hard error. */
        info->result = BLOCK_CLIENT_FAILED;
        return -1;
    } else if (rsp.status == STATUS_INTERNAL_SERVER_ERROR) {
        seaf_warning ("Internal server error.\n");
        info->result = BLOCK_CLIENT_SERVER_ERROR;
        return -1;
    }

    seaf_warning ("Bad status code %d in handshake.\n", rsp.status);
    info->result = BLOCK_CLIENT_FAILED;
    return -1;
}

static int
transfer_next_block (BlockTxClient *client)
{
    TransferTask *task = client->info->task;

    if (client->curr_block_id) {
        g_queue_pop_head (task->block_ids);
        g_free (client->curr_block_id);
        client->curr_block_id = NULL;
    }

    if (g_queue_get_length (task->block_ids) == 0) {
        seaf_debug ("Transfer blocks done.\n");
        client->info->result = BLOCK_CLIENT_SUCCESS;
        client->break_loop = TRUE;
        return 0;
    }

    client->curr_block_id = g_queue_peek_head (task->block_ids);

    if (task->type == TASK_TYPE_UPLOAD) {
        seaf_debug ("Put block %s.\n", client->curr_block_id);

        if (send_block_header (client, REQUEST_COMMAND_PUT) < 0) {
            seaf_warning ("Failed to send block header for PUT %s.\n",
                          client->curr_block_id);
            return -1;
        }

        if (send_block_content (client) < 0) {
            seaf_warning ("Failed to send block content for %s.\n",
                          client->curr_block_id);
            return -1;
        }

        seaf_debug ("recv_state set to HEADER.\n");

        client->parser.content_cb = handle_block_header_content_cb;
        client->recv_state = RECV_STATE_HEADER;
    } else {
        seaf_debug ("Get block %s.\n", client->curr_block_id);

        if (send_block_header (client, REQUEST_COMMAND_GET) < 0) {
            seaf_warning ("Failed to send block header for GET %s.\n",
                          client->curr_block_id);
            return -1;
        }

        seaf_debug ("recv_state set to HEADER.\n");

        client->parser.content_cb = handle_block_header_content_cb;
        client->parser.fragment_cb = save_block_content_cb;
        client->recv_state = RECV_STATE_HEADER;
    }

    return 0;
}

static FSEventStreamRef
add_watch (SeafWTMonitor *monitor, const char* repo_id, const char* worktree)
{
    SeafWTMonitorPriv *priv = monitor->priv;
    RepoWatchInfo *info;
    double latency = 0.25; /* unit: second */

    char *worktree_nfd = g_utf8_normalize (worktree, -1, G_NORMALIZE_NFD);

    CFStringRef mypaths[1];
    mypaths[0] = CFStringCreateWithCString (kCFAllocatorDefault,
                                            worktree_nfd, kCFStringEncodingUTF8);
    g_free (worktree_nfd);
    CFArrayRef pathsToWatch = CFArrayCreate(NULL, (const void **)mypaths, 1, NULL);
    FSEventStreamRef stream;

    /* Create the stream, passing in a callback */
    seaf_debug("Use kFSEventStreamCreateFlagWatchRoot\n");
    // kFSEventStreamCreateFlagFileEvents does not work for libraries with name
    // containing accent characters.
    struct FSEventStreamContext ctx = {0, monitor, NULL, NULL, NULL};
    stream = FSEventStreamCreate(kCFAllocatorDefault,
                                 stream_callback,
                                 &ctx,
                                 pathsToWatch,
                                 kFSEventStreamEventIdSinceNow,
                                 latency,
                                 kFSEventStreamCreateFlagWatchRoot
                                 );

    CFRelease (mypaths[0]);
    CFRelease (pathsToWatch);

    if (!stream) {
        seaf_warning ("[wt] Failed to create event stream.\n");
        return stream;
    }

    FSEventStreamScheduleWithRunLoop(stream, CFRunLoopGetCurrent(), kCFRunLoopDefaultMode);
    FSEventStreamStart (stream);
    /* FSEventStreamShow (stream); */
    seaf_debug ("[wt mon] Add repo %s watch success: %s.\n", repo_id, worktree);

    pthread_mutex_lock (&priv->hash_lock);
    g_hash_table_insert (priv->handle_hash,
                         g_strdup(repo_id), (gpointer)(long)stream);

    info = create_repo_watch_info (repo_id, worktree);
    g_hash_table_insert (priv->info_hash, (gpointer)(long)stream, info);
    pthread_mutex_unlock (&priv->hash_lock);

    /* An empty path indicates repo-mgr to scan the whole worktree. */
    add_event_to_queue (info->status, WT_EVENT_CREATE_OR_UPDATE, "", NULL);
    return stream;
}

static int
handle_handshake_request (BlockTxServer *server)
{
    HandshakeRequest req;
    struct evbuffer *input = server->recv_buf;
    unsigned char *enc_session_key;

    if (!server->session_key_len) {
        if (evbuffer_get_length (input) < sizeof(req))
            return 0;

        evbuffer_remove (input, &req, sizeof(req));

        req.version = ntohl (req.version);
        server->version = MIN (req.version, BLOCK_PROTOCOL_VERSION);
        if (server->version != 1 && server->version != 2) {
            seaf_warning ("Bad block protocol version %d.\n", server->version);
            send_handshake_response (server, STATUS_VERSION_MISMATCH);
            return -1;
        }

        seaf_debug ("Block protocol version %d.\n", server->version);

        server->session_key_len = ntohl (req.key_len);
        if (server->session_key_len > MAX_SESSION_KEY_SIZE) {
            seaf_warning ("Encrypted session key is too long: %d.\n",
                          server->session_key_len);
            send_handshake_response (server, STATUS_BAD_REQUEST);
            return -1;
        }
    }

    if (evbuffer_get_length (input) < server->session_key_len)
        return 0;

    enc_session_key = g_malloc (server->session_key_len);

    evbuffer_remove (input, enc_session_key, server->session_key_len);

    if (process_session_key (server, enc_session_key) < 0) {
        g_free (enc_session_key);
        return -1;
    }
    g_free (enc_session_key);

    if (send_handshake_response (server, STATUS_OK) < 0)
        return -1;

    seaf_debug ("recv_state set to AUTH.\n");

    server->parser.content_cb = handle_auth_req_content_cb;
    server->recv_state = RECV_STATE_AUTH;

    return 0;
}

static void
process_one_event (const char* eventPath,
                   RepoWatchInfo *info,
                   const char *worktree,
                   const FSEventStreamEventId eventId,
                   const FSEventStreamEventFlags eventFlags)
{
    WTStatus *status = info->status;
    char *filename;
    const char *tmp;
    tmp = eventPath + strlen(worktree);
    if (*tmp == '/')
        tmp++;
    filename = g_strdup(tmp);

    /* Reinterpreted RENAMED as combine of CREATED or DELETED event */
    if (eventFlags & kFSEventStreamEventFlagItemRenamed) {
        seaf_debug ("Rename Event Affected: %s \n", filename);
        struct stat buf;
        if (stat (eventPath, &buf)) {
            /* ret = -1, file is gone */
            add_event_to_queue (status, WT_EVENT_DELETE, filename, NULL);
        } else {
            /* ret = 0, file is here, but rename behaviour is unknown to us */
            add_event_to_queue (status, WT_EVENT_CREATE_OR_UPDATE, filename, NULL);
        }
    } else if (eventFlags & kFSEventStreamEventFlagItemModified) {
        seaf_debug ("Modified %s.\n", filename);
        add_event_to_queue (status, WT_EVENT_CREATE_OR_UPDATE, filename, NULL);
    } else if (eventFlags & kFSEventStreamEventFlagItemCreated) {
        seaf_debug ("Created %s.\n", filename);
         /**
          * no need to rechecking recursively in FSEventStream
          *
          * these flags are useful if necessary:
          * kFSEventStreamEventFlagItemIsFile
          * kFSEventStreamEventFlagItemIsDir
          * kFSEventStreamEventFlagItemIsSymlink
          */
        add_event_to_queue (status, WT_EVENT_CREATE_OR_UPDATE, filename, NULL);
    } else if (eventFlags & kFSEventStreamEventFlagItemRemoved) {
        seaf_debug ("Deleted %s.\n", filename);
        add_event_to_queue (status, WT_EVENT_DELETE, filename, NULL);
    } else if (eventFlags & kFSEventStreamEventFlagItemXattrMod) {
        seaf_debug ("XattrMod %s.\n", filename);
        add_event_to_queue (status, WT_EVENT_ATTRIB, filename, NULL);
    }
    //TODO: kFSEventStreamEventFlagRootChanged and
    //kFSEventStreamCreateFlagWatchRoot
    g_free (filename);
    g_atomic_int_set (&info->status->last_changed, (gint)time(NULL));
}

static int
schedule_merge_tasks (void *vscheduler)
{
    MergeScheduler *scheduler = vscheduler;
    int n_running = g_hash_table_size (scheduler->running);
    MergeTask *task;

    /* seaf_debug ("Waiting tasks %d, running tasks %d.\n", */
    /*             g_queue_get_length (scheduler->queue), n_running); */

    if (n_running >= MAX_RUNNING_TASKS)
        return TRUE;

    pthread_mutex_lock (&scheduler->q_lock);

    while (n_running < MAX_RUNNING_TASKS) {
        task = g_queue_pop_head (scheduler->queue);
        if (!task)
            break;

        if (!g_hash_table_lookup (scheduler->running, task->repo_id)) {
            int ret = ccnet_job_manager_schedule_job (scheduler->tpool,
                                                      merge_virtual_repo,
                                                      merge_virtual_repo_done,
                                                      task);
            if (ret < 0) {
                g_queue_push_tail (scheduler->queue, task);
                break;
            }

            g_hash_table_insert (scheduler->running,
                                 g_strdup(task->repo_id),
                                 task);
            n_running++;

            seaf_debug ("Run task for repo %.8s.\n", task->repo_id);
        } else {
            seaf_debug ("A task for repo %.8s is already running.\n", task->repo_id);

            g_queue_push_tail (scheduler->queue, task);
            break;
        }
    }

    pthread_mutex_unlock (&scheduler->q_lock);

    return TRUE;
}

static int
send_authentication (BlockTxClient *client)
{
    TransferTask *task = client->info->task;
    EVP_CIPHER_CTX ctx;
    int ret = 0;

    if (client->version == 1)
        blocktx_encrypt_init (&ctx, client->key, client->iv);
    else if (client->version == 2)
        blocktx_encrypt_init (&ctx, client->key_v2, client->iv_v2);

    seaf_debug ("session token length is %d.\n", strlen(task->session_token));

    if (send_encrypted_data_frame_begin (client->data_fd,
                                         strlen(task->session_token) + 1) < 0) {
        seaf_warning ("Send auth request: failed to begin.\n");
        client->info->result = BLOCK_CLIENT_NET_ERROR;
        ret = -1;
        goto out;
    }

    if (send_encrypted_data (&ctx, client->data_fd,
                             task->session_token,
                             strlen(task->session_token) + 1) < 0)
    {
        seaf_warning ("Send auth request: failed to send data.\n");
        client->info->result = BLOCK_CLIENT_NET_ERROR;
        ret = -1;
        goto out;
    }

    if (send_encrypted_data_frame_end (&ctx, client->data_fd) < 0) {
        seaf_warning ("Send auth request: failed to end.\n");
        client->info->result = BLOCK_CLIENT_NET_ERROR;
        ret = -1;
        goto out;
    }

    seaf_debug ("recv_state set to AUTH.\n");

    client->parser.content_cb = handle_auth_rsp_content_cb;
    client->recv_state = RECV_STATE_AUTH;

out:
    EVP_CIPHER_CTX_cleanup (&ctx);
    return ret;
}

static void
handle_response (CcnetProcessor *processor,
                 char *code, char *code_msg,
                 char *content, int clen)
{
    SeafileSendfsProc *proc = (SeafileSendfsProc *)processor;
    TransferTask *task = proc->tx_task;

    switch (processor->state) {
    case SEND_ROOT:
        if (strncmp(code, SC_OK, 3) == 0) {
            send_fs_roots (processor);
            return;
        }
        break;
    case SEND_OBJECT:
        if (strncmp(code, SC_GET_OBJECT, 3) == 0) {
            send_fs_objects (processor, content, clen);
            return;
        } else if (strncmp(code, SC_END, 3) == 0) {
            seaf_debug ("Send fs objects end.\n");
            ccnet_processor_done (processor, TRUE);
            return;
        }
        break;
    default:
        g_assert (0);
    }

    g_warning ("Bad response: %s %s.\n", code, code_msg);
    if (memcmp (code, SC_ACCESS_DENIED, 3) == 0)
        transfer_task_set_error (task, TASK_ERR_ACCESS_DENIED);
    ccnet_processor_done (processor, FALSE);
}

static void
send_commit (CcnetProcessor *processor, const char *object_id)
{
    char *data;
    int len;
    ObjectPack *pack = NULL;
    int pack_size;

    if (seaf_obj_store_read_obj (seaf->commit_mgr->obj_store,
                                 object_id, (void**)&data, &len) < 0) {
        g_warning ("Failed to read commit %s.\n", object_id);
        goto fail;
    }

    pack_size = sizeof(ObjectPack) + len;
    pack = malloc (pack_size);
    memcpy (pack->id, object_id, 41);
    memcpy (pack->object, data, len);

    ccnet_processor_send_update (processor, SC_OBJECT, SS_OBJECT,
                                 (char *)pack, pack_size);

    seaf_debug ("Send commit %.8s.\n", object_id);

    g_free (data);
    free (pack);
    return;

fail:
    ccnet_processor_send_update (processor, SC_NOT_FOUND, SS_NOT_FOUND,
                                 object_id, 41);
    ccnet_processor_done (processor, FALSE);
}

static void
on_seafdir_read (OSAsyncResult *res, void *cb_data)
{
    CcnetProcessor *processor = cb_data;
    SeafDir *dir;
    USE_PRIV;

    --(priv->inspect_objects);
    --(priv->checking_dirs);

    if (!res->success) {
        request_object_batch (processor, priv, res->obj_id);
        return;
    }

#ifdef DEBUG
    seaf_debug ("[recvfs] Read seafdir %s.\n", res->obj_id);
#endif

    dir = seaf_dir_from_data (res->obj_id, res->data, res->len,
                              (priv->repo_version > 0));
    if (!dir) {
        seaf_warning ("[recvfs] Corrupt dir object %s.\n", res->obj_id);
        request_object_batch (processor, priv, res->obj_id);
        return;
    }

    int ret = check_seafdir (processor, dir);
    seaf_dir_free (dir);
    if (ret < 0)
        return;
}

static void
handle_update (CcnetProcessor *processor,
               char *code, char *code_msg,
               char *content, int clen)
{
    switch (processor->state) {
    case SEND_OBJECTS:
        if (strncmp (code, SC_OBJ_LIST_SEG, 3) == 0) {
            process_object_list_segment (processor, content, clen);
            return;
        } else if (strncmp (code, SC_END, 3) == 0) {
            seaf_debug ("All objects received. Done.\n");
            ccnet_processor_done (processor, TRUE);
            return;
        }
        break;
    default:
        g_return_if_reached ();
    }

    seaf_warning ("Bad update: %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
read_done_cb (OSAsyncResult *res, void *cb_data)
{
    CcnetProcessor *processor = cb_data;
    USE_PRIV;

    if (!res->success) {
        g_warning ("[putcommit] Failed to read %s.\n", res->obj_id);
        goto bad;
    }

    send_commit (processor, res->obj_id, res->data, res->len);

    seaf_debug ("Send commit %.8s.\n", res->obj_id);

    /* Send next commit. */
    if (priv->id_list != NULL)
        read_and_send_commit (processor);
    else {
        ccnet_processor_send_response (processor, SC_END, SS_END, NULL, 0);
        ccnet_processor_done (processor, TRUE);
    }

    return;

bad:
    ccnet_processor_send_response (processor, SC_NOT_FOUND, SS_NOT_FOUND,
                                   NULL, 0);
    ccnet_processor_done (processor, FALSE);
}

int
seaf_wt_monitor_refresh_repo (SeafWTMonitor *monitor, const char *repo_id)
{
    WatchCommand cmd;
    int res;

    memset (&cmd, 0, sizeof(cmd));
    memcpy (cmd.repo_id, repo_id, 37);
    cmd.type = CMD_REFRESH_WATCH;

    int n = seaf_pipe_writen (monitor->cmd_pipe[1], &cmd, sizeof(cmd));

    if (n != sizeof(cmd)) {
        seaf_warning ("[wt mon] fail to write command pipe.\n");
        return -1;
    }

    seaf_debug ("send a refresh command, repo %s\n", repo_id);

    n = seaf_pipe_readn (monitor->res_pipe[0], &res, sizeof(int));
    if (n != sizeof(int)) {
        seaf_warning ("[wt mon] fail to read result pipe.\n");
        return -1;
    }

    return res;
}

static void
process_object_list_segment (CcnetProcessor *processor, char *content, int clen)
{
    int n, i;
    char *p;

    if (clen % 40 != 0) {
        seaf_warning ("Invalid object list segment length %d.\n", clen);
        ccnet_processor_send_response (processor, SC_SHUTDOWN, SS_SHUTDOWN, NULL, 0);
        ccnet_processor_done (processor, FALSE);
        return;
    }

    n = clen/40;
    p = content;

    seaf_debug ("%d objects are needed by the client.\n", n);

    char *obj_id;
    for (i = 0; i < n; ++i) {
        obj_id = g_strndup(p, 40);
        read_fs_object (processor, obj_id);
        g_free (obj_id);
        p += 40;
    }
}

int
seaf_wt_monitor_watch_repo (SeafWTMonitor *monitor,
                            const char *repo_id,
                            const char *worktree)
{
    WatchCommand cmd;
    int res;

    memset (&cmd, 0, sizeof(cmd));
    memcpy (cmd.repo_id, repo_id, 37);
    cmd.type = CMD_ADD_WATCH;
    g_strlcpy (cmd.worktree, worktree, SEAF_PATH_MAX);

    int n = seaf_pipe_writen (monitor->cmd_pipe[1], &cmd, sizeof(cmd));
    
    if (n != sizeof(cmd)) {
        seaf_warning ("[wt mon] fail to write command pipe.\n");
        return -1;
    }

    seaf_debug ("send a watch command, repo %s\n", repo_id);

    n = seaf_pipe_readn (monitor->res_pipe[0], &res, sizeof(int));
    if (n != sizeof(int)) {
        seaf_warning ("[wt mon] fail to read result pipe.\n");
        return -1;
    }

    return res;
}