/* Code for JSON_RPC API.
 *
 * Each socket connection is represented by a `struct json_connection`.
 *
 * This can have zero, one or more `struct command` in progress at a time:
 * because the json_connection can be closed at any point, these `struct command`
 * have a independent lifetimes.
 *
 * Each `struct command` writes into a `struct json_stream`, which is created
 * the moment they start writing output (see attach_json_stream).  Initially
 * the struct command owns it since they're writing into it.  When they're
 * done, the `json_connection` needs to drain it (if it's still around).  At
 * that point, the `json_connection` becomes the owner (or it's simply freed).
 */
/* eg: { "jsonrpc":"2.0", "method" : "dev-echo", "params" : [ "hello", "Arabella!" ], "id" : "1" } */
#include "config.h"
#include <ccan/array_size/array_size.h>
#include <ccan/asort/asort.h>
#include <ccan/err/err.h>
#include <ccan/io/io.h>
#include <ccan/json_escape/json_escape.h>
#include <ccan/json_out/json_out.h>
#include <ccan/tal/path/path.h>
#include <ccan/tal/str/str.h>
#include <common/codex32.h>
#include <common/json_command.h>
#include <common/json_filter.h>
#include <common/jsonrpc_io.h>
#include <common/memleak.h>
#include <common/timeout.h>
#include <common/trace.h>
#include <db/common.h>
#include <db/exec.h>
#include <errno.h>
#include <fcntl.h>
#include <lightningd/jsonrpc.h>
#include <lightningd/plugin_hook.h>


/* Dummy structure. */
struct command_result {
	char c;
};

static struct command_result param_failed, complete, pending, unknown;

struct command_result *command_param_failed(void)
{
	return &param_failed;
}

/* We immediately respond with success if we reach here. */
struct command_result *command_check_done(struct command *cmd)
{
	struct json_stream *response;

	assert(cmd->mode == CMD_CHECK);

	response = json_stream_success(cmd);
	json_add_string(response, "command_to_check", cmd->json_cmd->name);
	return command_success(cmd, response);
}

struct command_result *command_its_complicated(const char *relationship_details
					       UNNEEDED)
{
	return &unknown;
}

/* This represents a JSON RPC connection.  It can invoke multiple commands, but
 * a command can outlive the connection, which could close any time. */
struct json_connection {
	/* The global state */
	struct lightningd *ld;

	/* This io_conn (and our owner!) */
	struct io_conn *conn;

	/* Logging for this json connection. */
	struct logger *log;

	/* The buffer and state reading in the JSON commands */
	struct jsonrpc_io *json_in;

	/* Local deprecated support? */
	bool deprecated_ok;

	/* Our commands */
	struct list_head commands;

	/* Are notifications enabled? */
	bool notifications_enabled;

	/* Are we allowed to batch database commitments? */
	bool db_batching;

	/* Our json_streams (owned by the commands themselves while running).
	 * Since multiple streams could start returning data at once, we
	 * always service these in order. */
	struct list_head jsouts;
};

/* We don't put usage inside struct json_command as it's good practice
 * to have those const. */
struct cmd_and_usage {
	const struct json_command *command;
	const char *usage;
};

/**
 * `jsonrpc` encapsulates the entire state of the JSON-RPC interface,
 * including a list of methods that the interface supports (can be
 * appended dynamically, e.g., for plugins, and logs. It also serves
 * as a convenient `tal`-parent for all JSON-RPC related allocations.
 */
struct jsonrpc {
	struct io_listener *rpc_listener;

	/* Can't be const: we set ->usage later */
	STRMAP(struct cmd_and_usage *) cmdmap;
};

/* The command itself usually owns the stream, because jcon may get closed.
 * The command transfers ownership once it's done though. */
static struct json_stream *jcon_new_json_stream(const tal_t *ctx,
						struct json_connection *jcon,
						struct command *writer)
{
	struct json_stream *js = new_json_stream(ctx, writer, jcon->log);

	/* Wake writer to start streaming, in case it's not already. */
	io_wake(jcon);
	list_add_tail(&jcon->jsouts, &js->list);
	return js;
}

/* jcon and cmd have separate lifetimes: we detach them on either destruction */
static void destroy_jcon(struct json_connection *jcon)
{
	struct command *c;

	list_for_each(&jcon->commands, c, list)
		c->jcon = NULL;

	/* Make sure this happens last! */
	tal_free(jcon->log);
}

struct logger *command_logger(struct command *cmd)
{
	if (cmd->jcon)
		return cmd->jcon->log;
	return cmd->ld->log;
}

static struct command_result *json_help(struct command *cmd,
					const char *buffer,
					const jsmntok_t *obj UNNEEDED,
					const jsmntok_t *params);

static const struct json_command help_command = {
	"help",
	json_help,
};
AUTODATA(json_command, &help_command);

/* We prepare a canned JSON response, for top level to write as reply
 * immediately before we exit. */
static struct command_result *prepare_stop_conn(struct command *cmd,
						const char *why)
{
	struct json_out *jout;
	const char *p;
	size_t len;

	/* With rpc_command_hook, jcon might have closed in the meantime! */
	if (!cmd->jcon) {
		/* Return us to toplevel lightningd.c */
		log_debug(cmd->ld->log, "io_break: %s", __func__);
		io_break(cmd->ld);
		return command_still_pending(cmd);
	}

	cmd->ld->stop_conn = cmd->jcon->conn;

	jout = json_out_new(tmpctx);
	json_out_start(jout, NULL, '{');
	json_out_addstr(jout, "jsonrpc", "2.0");
	/* Copy input id token exactly */
	memcpy(json_out_member_direct(jout, "id", strlen(cmd->id)),
	       cmd->id, strlen(cmd->id));
	json_out_start(jout, "result", '{');
	json_out_addstr(jout, "result", why);
	json_out_end(jout, '}');
	json_out_end(jout, '}');
 	json_out_finished(jout);

	/* Add two \n */
	memcpy(json_out_direct(jout, 2), "\n\n", strlen("\n\n"));
	p = json_out_contents(jout, &len);
	cmd->ld->stop_response = tal_strndup(cmd->ld, p, len);

	/* Wake write loop in case it's not already. */
	io_wake(cmd->jcon);

	return command_still_pending(cmd);
}

static struct command_result *json_stop(struct command *cmd,
					const char *buffer,
					const jsmntok_t *obj UNNEEDED,
					const jsmntok_t *params)
{
	if (!param(cmd, buffer, params, NULL))
		return command_param_failed();

	log_unusual(cmd->ld->log, "JSON-RPC shutdown");
	return prepare_stop_conn(cmd, "Shutdown complete");
}

static const struct json_command stop_command = {
	"stop",
	json_stop,
};
AUTODATA(json_command, &stop_command);

static bool have_channels(struct lightningd *ld)
{
	struct peer_node_id_map_iter it;
	struct peer *peer;

	for (peer = peer_node_id_map_first(ld->peers, &it);
	     peer;
	     peer = peer_node_id_map_next(ld->peers, &it)) {
		if (peer->uncommitted_channel)
			return true;
		if (!list_empty(&peer->channels))
			return true;
	}
	return false;
}

static struct command_result *param_hsm_secret(struct command *cmd,
					       const char *name,
					       const char *buffer,
					       const jsmntok_t *tok,
					       const char **hsm_secret)
{
	char *err;
	/* We parse here for sanity checking, but we just hand string to --recover */
	const struct hsm_secret *hsms;

	*hsm_secret = json_strdup(cmd, buffer, tok);
	err = hsm_secret_arg(tmpctx, *hsm_secret, &hsms);
	if (err)
		return command_fail_badparam(cmd, name, buffer, tok, err);
	return NULL;
}

/* We cannot --recover unless these files are not in place. */
static void move_prerecover_files(const char *dir)
{
	const char *files[] = {
		"lightningd.sqlite3",
		"emergency.recover",
		"hsm_secret",
	};

	if (mkdir(dir, 0770) != 0)
		fatal("Could not make %s: %s", dir, strerror(errno));
	for (size_t i = 0; i < ARRAY_SIZE(files); i++) {
		if (rename(files[i], path_join(tmpctx, dir, files[i])) != 0) {
			fatal("Could not move %s: %s", files[i], strerror(errno));
		}
	}
}

static struct command_result *json_recover(struct command *cmd,
					   const char *buffer,
					   const jsmntok_t *obj UNNEEDED,
					   const jsmntok_t *params)
{
	const char *dir, *hsm_secret;

	if (!param_check(cmd, buffer, params,
			 p_req("hsmsecret", param_hsm_secret, &hsm_secret),
			 NULL))
		return command_param_failed();

	/* FIXME: How do we "move" the Postgres DB? */
	if (!streq(cmd->ld->wallet->db->config->name, "sqlite3"))
		return command_fail(cmd, LIGHTNINGD,
				    "Only sqlite3 supported for recover command");

	/* Check this is an empty node! */
	if (db_get_intvar(cmd->ld->wallet->db, "bip32_max_index", 0) != 0
	    || db_get_intvar(cmd->ld->wallet->db, "bip86_max_index", 0) != 0) {
		return command_fail(cmd, RECOVER_NODE_IN_USE,
				    "Node has already issued bitcoin addresses!");
	}

	if (have_channels(cmd->ld)) {
		return command_fail(cmd, RECOVER_NODE_IN_USE,
				    "Node has channels!");
	}

	/* Don't try to add --recover to cmdline twice! */
	if (cmd->ld->recover != NULL) {
		return command_fail(cmd, RECOVER_NODE_IN_USE,
				    "Already doing recover");
	}

	if (command_check_only(cmd))
		return command_check_done(cmd);

	dir = tal_fmt(tmpctx, "lightning.pre-recover.%u", getpid());
	log_unusual(cmd->ld->log,
		    "JSON-RPC recovery command: moving existing files to %s", dir);

	move_prerecover_files(dir);

	/* Top level with add --recover=... here */
	cmd->ld->recover_secret = tal_steal(cmd->ld, hsm_secret);
	cmd->ld->try_reexec = true;
	return prepare_stop_conn(cmd, "Recovery restart in progress");
}

static const struct json_command recover_command = {
	"recover",
	json_recover,
};
AUTODATA(json_command, &recover_command);

struct slowcmd {
	struct command *cmd;
	unsigned *msec;
	struct json_stream *js;
};

static void slowcmd_finish(struct slowcmd *sc)
{
	json_add_num(sc->js, "msec", *sc->msec);
	was_pending(command_success(sc->cmd, sc->js));
}

static void slowcmd_start(struct slowcmd *sc)
{
	sc->js = json_stream_success(sc->cmd);
	new_reltimer(sc->cmd->ld->timers, sc, time_from_msec(*sc->msec),
		     slowcmd_finish, sc);
}

static struct command_result *json_dev(struct command *cmd UNUSED,
				       const char *buffer,
				       const jsmntok_t *obj UNNEEDED,
				       const jsmntok_t *params)
{
	const char *subcmd;

	subcmd = param_subcommand(cmd, buffer, params,
				  "crash", "rhash", "slowcmd", NULL);
	if (!subcmd)
		return command_param_failed();

	if (streq(subcmd, "crash")) {
		if (!param(cmd, buffer, params,
			   p_req("subcommand", param_ignore, cmd),
			   NULL))
			return command_param_failed();
		fatal("Crash at user request");
	} else if (streq(subcmd, "slowcmd")) {
		struct slowcmd *sc = tal(cmd, struct slowcmd);

		sc->cmd = cmd;
		if (!param(cmd, buffer, params,
			   p_req("subcommand", param_ignore, cmd),
			   p_opt_def("msec", param_number, &sc->msec, 1000),
			   NULL))
			return command_param_failed();

		new_reltimer(cmd->ld->timers, sc, time_from_msec(0),
			     slowcmd_start, sc);
		return command_still_pending(cmd);
	} else {
		assert(streq(subcmd, "rhash"));
		struct json_stream *response;
		struct sha256 *secret;

		if (!param(cmd, buffer, params,
			   p_req("subcommand", param_ignore, cmd),
			   p_req("secret", param_sha256, &secret),
			   NULL))
			return command_param_failed();

		/* Hash in place. */
		sha256(secret, secret, sizeof(*secret));
		response = json_stream_success(cmd);
		json_add_sha256(response, "rhash", secret);
		return command_success(cmd, response);
	}
}

static const struct json_command dev_command = {
	"dev",
	json_dev,
	.dev_only = true,
};
AUTODATA(json_command, &dev_command);

static struct json_command **get_cmdlist(size_t *num_cmdlist)
{
	static struct json_command **cmdlist;
	if (!cmdlist)
		cmdlist = autodata_get(json_command, num_cmdlist);

	return cmdlist;
}

struct json_help_info {
	struct command *cmd;
	struct json_stream *response;
};

/* Used as a strmap_iterate function: returns true to continue */
static bool json_add_help_command(const char *cmdname,
				  struct cmd_and_usage *cmd,
				  struct json_help_info *hinfo)
{
	char *usage;

	/* If they disallow deprecated APIs, don't even list them */
	if (!command_deprecated_out_ok(hinfo->cmd, NULL,
				       cmd->command->depr_start,
				       cmd->command->depr_end)) {
		return true;
	}

	usage = tal_fmt(tmpctx, "%s%s %s",
			cmd->command->name,
			cmd->command->depr_start ? " (DEPRECATED!)" : "",
			cmd->usage);
	json_object_start(hinfo->response, NULL);
	json_add_string(hinfo->response, "command", usage);
	json_object_end(hinfo->response);
	return true;
}

static struct command_result *json_help(struct command *cmd,
					const char *buffer,
					const jsmntok_t *obj UNNEEDED,
					const jsmntok_t *params)
{
	const char *cmdname;
	struct cmd_and_usage *one_cmd;
	struct json_help_info hinfo;

	if (!param_check(cmd, buffer, params,
			 p_opt("command", param_string, &cmdname),
			 NULL))
		return command_param_failed();

	if (cmdname) {
		one_cmd = strmap_get(&cmd->ld->jsonrpc->cmdmap, cmdname);
		if (!one_cmd)
			return command_fail(cmd, JSONRPC2_METHOD_NOT_FOUND,
					    "Unknown command %s",
					    cmdname);
		if (!command_deprecated_in_ok(cmd, NULL,
					      one_cmd->command->depr_start,
					      one_cmd->command->depr_end))
			return command_fail(cmd, JSONRPC2_METHOD_NOT_FOUND,
					    "Deprecated command %s",
					    cmdname);
		if (!cmd->ld->developer && one_cmd->command->dev_only)
			return command_fail(cmd, JSONRPC2_METHOD_NOT_FOUND,
					    "Developer-only command %s",
					    cmdname);
	} else
		one_cmd = NULL;

	if (command_check_only(cmd))
		return command_check_done(cmd);

	hinfo.cmd = cmd;
	hinfo.response = json_stream_success(cmd);
	json_array_start(hinfo.response, "help");
	if (one_cmd)
		json_add_help_command(cmdname, one_cmd, &hinfo);
	else {
		strmap_iterate(&cmd->ld->jsonrpc->cmdmap,
			       json_add_help_command, &hinfo);
	}
	json_array_end(hinfo.response);

	/* Tell cli this is simple enough to be formatted flat for humans */
	json_add_string(hinfo.response, "format-hint", "simple");

	return command_success(cmd, hinfo.response);
}

static const struct json_command *find_cmd(const struct jsonrpc *rpc,
					   const char *buffer,
					   const jsmntok_t *tok)
{
	const struct cmd_and_usage *cmd;

	cmd = strmap_getn(&rpc->cmdmap, buffer + tok->start, tok->end - tok->start);
	if (cmd)
		return cmd->command;
	return NULL;
}

/* This can be called directly on shutdown, even with unfinished cmd */
static void destroy_command(struct command *cmd)
{
	if (!cmd->jcon) {
		log_debug(cmd->ld->log,
			    "Command returned result after jcon close");
		return;
	}
	list_del_from(&cmd->jcon->commands, &cmd->list);
}

struct command_result *command_raw_complete(struct command *cmd,
					    struct json_stream *result)
{
	json_stream_close(result, cmd);

	/* If we have a jcon, it will free result for us. */
	if (cmd->jcon)
		tal_steal(cmd->jcon, result);

	tal_free(cmd);
	return &complete;
}

struct command_result *command_success(struct command *cmd,
				       struct json_stream *result)
{
	assert(cmd);
	assert(cmd->json_stream == result);

	/* Filter will get upset if we close "result" object it didn't
	 * see! */
	if (cmd->filter) {
		const char *err = json_stream_detach_filter(tmpctx, result);
		if (err)
			json_add_string(result, "warning_parameter_filter",
					err);
	}

	json_object_end(result);
	json_object_end(result);

	return command_raw_complete(cmd, result);
}

struct command_result *command_failed(struct command *cmd,
				      struct json_stream *result)
{
	assert(cmd->json_stream == result);
	/* Have to close error */
	json_object_end(result);
	json_object_end(result);

	return command_raw_complete(cmd, result);
}

struct command_result *command_fail(struct command *cmd, enum jsonrpc_errcode code,
				    const char *fmt, ...)
{
	const char *errmsg;
	struct json_stream *r;
	va_list ap;

	va_start(ap, fmt);
	errmsg = tal_vfmt(cmd, fmt, ap);
	va_end(ap);
	r = json_stream_fail_nodata(cmd, code, errmsg);

	return command_failed(cmd, r);
}

struct json_filter **command_filter_ptr(struct command *cmd)
{
	return &cmd->filter;
}

bool command_deprecated_ok_flag(const struct command *cmd)
{
	if (cmd->jcon)
		return cmd->jcon->deprecated_ok;
	return cmd->ld->deprecated_ok;
}

bool command_deprecated_in_ok(struct command *cmd,
				 const char *param,
				 const char *depr_start,
				 const char *depr_end)
{
	return lightningd_deprecated_in_ok(cmd->ld,
					   command_logger(cmd),
					   command_deprecated_ok_flag(cmd),
					   cmd->json_cmd->name, param,
					   depr_start, depr_end,
					   cmd->id);
}

bool command_deprecated_out_ok(struct command *cmd,
			       const char *fieldname,
			       const char *depr_start,
			       const char *depr_end)
{
	return lightningd_deprecated_out_ok(cmd->ld,
					    command_deprecated_ok_flag(cmd),
					    cmd->json_cmd->name,
					    fieldname,
					    depr_start, depr_end);
}

struct command_result *command_still_pending(struct command *cmd)
{
	notleak_with_children(cmd);
	cmd->pending = true;

	/* If we've started writing, wake reader. */
	if (cmd->json_stream)
		json_stream_flush(cmd->json_stream);

	return &pending;
}

static void json_command_malformed(struct json_connection *jcon,
				   const char *id,
				   const char *error)
{
	/* NULL writer is OK here, since we close it immediately. */
	struct json_stream *js = jcon_new_json_stream(jcon, jcon, NULL);

	json_object_start(js, NULL);
	json_add_string(js, "jsonrpc", "2.0");
	json_add_primitive(js, "id", id);
	json_object_start(js, "error");
	json_add_jsonrpc_errcode(js, "code", JSONRPC2_INVALID_REQUEST);
	json_add_string(js, "message", error);
	json_object_end(js);
	json_object_end(js);

	json_stream_close(js, NULL);
}

void json_notify_fmt(struct command *cmd,
		     enum log_level level,
		     const char *fmt, ...)
{
	va_list ap;
	struct json_stream *js;
	const char *msg;

	va_start(ap, fmt);
	msg = tal_vfmt(tmpctx, fmt, ap);
	va_end(ap);

	/* Help for tests */
	log_debug(cmd->ld->log, "NOTIFY %s %s %s",
		  cmd->id, log_level_name(level), msg);
	if (!cmd->send_notifications)
		return;

	js = json_stream_raw_for_cmd(cmd);

	json_object_start(js, NULL);
	json_add_string(js, "jsonrpc", "2.0");
	json_add_string(js, "method", "message");
	json_object_start(js, "params");
	json_add_id(js, cmd->id);
	json_add_string(js, "level", log_level_name(level));
	json_add_string(js, "message", msg);
	json_object_end(js);
	json_object_end(js);

	json_stream_double_cr(js);
	json_stream_flush(js);
}

struct json_stream *json_stream_raw_for_cmd(struct command *cmd)
{
	struct json_stream *js;

	/* Might have already opened it for a notification */
	if (cmd->json_stream)
		return cmd->json_stream;

	/* If they still care about the result, attach it to them. */
	if (cmd->jcon)
		js = jcon_new_json_stream(cmd, cmd->jcon, cmd);
	else
		js = new_json_stream(cmd, cmd, NULL);

	assert(!cmd->json_stream);
	cmd->json_stream = js;
	return js;
}

void json_stream_log_suppress_for_cmd(struct json_stream *js,
				     const struct command *cmd)
{
	const char *nm = cmd->json_cmd->name;
	const char *s = tal_fmt(tmpctx, "Suppressing logging of %s command", nm);
	log_io(cmd->jcon->log, LOG_IO_OUT, NULL, s, NULL, 0);

	/* Really shouldn't be used for anything else */
	assert(streq(nm, "getlog"));
	js->log = NULL;
}

static struct json_stream *json_start(struct command *cmd)
{
	struct json_stream *js = json_stream_raw_for_cmd(cmd);

	json_object_start(js, NULL);
	json_add_string(js, "jsonrpc", "2.0");
	json_add_id(js, cmd->id);
	return js;
}

struct json_stream *json_stream_success(struct command *cmd)
{
	struct json_stream *r = json_start(cmd);
	json_object_start(r, "result");

	/* We have results?  OK, start filtering */
	if (cmd->filter)
		json_stream_attach_filter(r, cmd->filter);
	return r;
}

struct json_stream *json_stream_fail_nodata(struct command *cmd,
					    enum jsonrpc_errcode code,
					    const char *errmsg)
{
	struct json_stream *js = json_start(cmd);

	assert(code);

	json_object_start(js, "error");
	json_add_jsonrpc_errcode(js, "code", code);
	json_add_string(js, "message", errmsg);

	return js;
}

struct json_stream *json_stream_fail(struct command *cmd,
				     enum jsonrpc_errcode code,
				     const char *errmsg)
{
	struct json_stream *r = json_stream_fail_nodata(cmd, code, errmsg);

	json_object_start(r, "data");
	return r;
}

static struct command_result *command_exec(struct json_connection *jcon,
                                           struct command *cmd,
                                           const char *buffer,
                                           const jsmntok_t *request,
                                           const jsmntok_t *params)
{
	struct command_result *res;

	res = cmd->json_cmd->dispatch(cmd, buffer, request, params);

	assert(res == &param_failed
	       || res == &complete
	       || res == &pending
	       || res == &unknown);

	/* If they didn't complete it, they must call command_still_pending.
	 * If they completed it, it's freed already. */
	if (res == &pending)
		assert(cmd->pending);

	return res;
}

/* A plugin hook to take over (fail/alter) RPC commands */
struct rpc_command_hook_payload {
	struct command *cmd;
	const char *buffer;
	const jsmntok_t *request;

	/* custom response/replace/error options plugins can have */
	const char *custom_result;
	const char *custom_error;
	const jsmntok_t *custom_replace;
	const char *custom_buffer;
};

static void rpc_command_hook_serialize(struct rpc_command_hook_payload *p,
				       struct json_stream *s,
				       struct plugin *plugin)
{
	const jsmntok_t *tok;
	size_t i;
	char *key;
	json_object_start(s, "rpc_command");

	json_for_each_obj(i, tok, p->request) {
		key = tal_strndup(NULL, p->buffer + tok->start,
				  tok->end - tok->start);
		json_add_tok(s, key, tok + 1, p->buffer);
		tal_free(key);
	}
	json_object_end(s);
}

static void replace_command(struct rpc_command_hook_payload *p,
			    const char *buffer,
			    const jsmntok_t *replacetok)
{
	const jsmntok_t *method = NULL, *params = NULL, *jsonrpc;
	const char *bad;

	/* Must contain "method", "params" and "id" */
	if (replacetok->type != JSMN_OBJECT) {
		bad = "'replace' must be an object";
		goto fail;
	}

	method = json_get_member(buffer, replacetok, "method");
	if (!method) {
		bad = "missing 'method'";
		goto fail;
	}
	params = json_get_member(buffer, replacetok, "params");
	if (!params) {
		bad = "missing 'params'";
		goto fail;
	}
	if (!json_get_member(buffer, replacetok, "id")) {
		bad = "missing 'id'";
		goto fail;
	}

	p->cmd->json_cmd = find_cmd(p->cmd->ld->jsonrpc, buffer, method);
	if (!p->cmd->json_cmd) {
		bad = tal_fmt(tmpctx, "redirected to unknown method '%.*s'",
			      method->end - method->start,
			      buffer + method->start);
		goto fail;
	}
	if (!command_deprecated_in_ok(p->cmd,
				      json_strdup(tmpctx, buffer, method),
				      p->cmd->json_cmd->depr_start,
				      p->cmd->json_cmd->depr_end)) {
		bad = tal_fmt(tmpctx, "redirected to deprecated command '%.*s'",
			      method->end - method->start,
			      buffer + method->start);
		goto fail;
	}
	if (p->cmd->json_cmd->dev_only && !p->cmd->ld->developer) {
		bad = tal_fmt(tmpctx, "redirected to developer-only command '%.*s'",
			      method->end - method->start,
			      buffer + method->start);
		goto fail;
	}

	jsonrpc = json_get_member(buffer, replacetok, "jsonrpc");
	if (!jsonrpc || jsonrpc->type != JSMN_STRING || !json_tok_streq(buffer, jsonrpc, "2.0")) {
		bad = "jsonrpc: \"2.0\" must be specified in the request";
		goto fail;
	}

	was_pending(command_exec(p->cmd->jcon, p->cmd, buffer, replacetok,
				 params));
	return;

fail:
	was_pending(command_fail(p->cmd, JSONRPC2_INVALID_REQUEST,
				 "Bad response to 'rpc_command' hook: %s (%.*s)",
				 bad,
				 json_tok_full_len(replacetok),
				 json_tok_full(buffer, replacetok)));
}

static void rpc_command_hook_final(struct rpc_command_hook_payload *p STEALS)
{
	const jsmntok_t *params;

	/* Free payload with cmd */
	tal_steal(p->cmd, p);

	if (p->custom_result != NULL) {
		struct json_stream *s = json_start(p->cmd);
		json_add_jsonstr(s, "result",
				 p->custom_result, strlen(p->custom_result));
		json_object_end(s);
		return was_pending(command_raw_complete(p->cmd, s));
	}
	if (p->custom_error != NULL) {
		struct json_stream *s = json_start(p->cmd);
		json_add_jsonstr(s, "error",
				 p->custom_error, strlen(p->custom_error));
		json_object_end(s);
		return was_pending(command_raw_complete(p->cmd, s));
	}
	if (p->custom_replace != NULL)
		return replace_command(p, p->custom_buffer, p->custom_replace);

	/* If no plugin requested a change, just continue command execution. */
	params = json_get_member(p->buffer, p->request, "params");
	return was_pending(command_exec(p->cmd->jcon,
					p->cmd,
					p->buffer,
					p->request,
					params));
}

static bool
rpc_command_hook_callback(struct rpc_command_hook_payload *p,
			  const char *buffer, const jsmntok_t *resulttok)
{
	const struct lightningd *ld = p->cmd->ld;
	const jsmntok_t *tok, *custom_return;
	static char *error = "";
	char *method;

	if (!resulttok || !buffer)
		return true;

	tok = json_get_member(buffer, resulttok, "result");
	if (tok) {
		if (!json_tok_streq(buffer, tok, "continue")) {
			error = "'result' should only be 'continue'.";
			goto log_error_and_skip;
		}
		/* plugin tells us to do nothing. just pass. */
		return true;
	}

	/* didn't just continue but hook was already modified by prior plugin */
	if (p->custom_result != NULL ||
	    p->custom_error != NULL ||
	    p->custom_replace != NULL) {
		/* get method name and log error (only the first time). */
		tok = json_get_member(p->buffer, p->request, "method");
		method = tal_strndup(p, p->buffer + tok->start, tok->end - tok->start);
		log_unusual(ld->log, "rpc_command hook '%s' already modified, ignoring.", method );
		rpc_command_hook_final(p);
		return false;
	}

	/* If the registered plugin did not respond with continue,
	 * it wants either to replace the request... */
	tok = json_get_member(buffer, resulttok, "replace");
	if (tok) {
		/* We need to make copies here, as buffer and tokens
		 * can be reused. */
		json_dup_contents(p, buffer, tok,
				  &p->custom_buffer, &p->custom_replace);
		return true;
	}

	/* ...or return a custom JSONRPC response. */
	tok = json_get_member(buffer, resulttok, "return");
	if (tok) {
		custom_return = json_get_member(buffer, tok, "result");
		if (custom_return) {
			p->custom_result = json_strdup(p, buffer, custom_return);
			return true;
		}

		custom_return = json_get_member(buffer, tok, "error");
		if (custom_return) {
			enum jsonrpc_errcode code;
			const char *errmsg;
			if (!json_to_jsonrpc_errcode(buffer,
						     json_get_member(buffer, custom_return, "code"),
						     &code)) {
				error = "'error' object does not contain a code.";
				goto log_error_and_skip;
			}
			errmsg = json_strdup(tmpctx, buffer,
			                     json_get_member(buffer, custom_return, "message"));
			if (!errmsg) {
				error = "'error' object does not contain a message.";
				goto log_error_and_skip;
			}
			p->custom_error = json_strdup(p, buffer, custom_return);
			return true;
		}
	}

log_error_and_skip:
	/* Just log BROKEN errors. Give other plugins a chance. */
	log_broken(ld->log, "Bad response to 'rpc_command' hook. %s", error);
	return true;
}

REGISTER_PLUGIN_HOOK_STRFILTER(rpc_command,
			       rpc_command_hook_callback,
			       rpc_command_hook_final,
			       rpc_command_hook_serialize,
			       struct rpc_command_hook_payload *);

/* We return struct command_result so command_fail return value has a natural
 * sink; we don't actually use the result. */
static struct command_result *
parse_request(struct json_connection *jcon,
	      const char *buffer,
	      const jsmntok_t tok[],
	      const char **methodname)
{
	const jsmntok_t *method, *id, *params, *filter, *jsonrpc;
	struct command *c;
	struct rpc_command_hook_payload *rpc_hook;
	bool completed;

	if (tok[0].type != JSMN_OBJECT) {
		json_command_malformed(jcon, "null",
				       "Expected {} for json command");
		return NULL;
	}

	method = json_get_member(buffer, tok, "method");
	params = json_get_member(buffer, tok, "params");
	filter = json_get_member(buffer, tok, "filter");
	id = json_get_member(buffer, tok, "id");

	if (!id) {
		json_command_malformed(jcon, "null", "No id");
		return NULL;
	}

	if (id->type != JSMN_STRING && id->type != JSMN_PRIMITIVE) {
		json_command_malformed(jcon, "null",
				       "Expected string/primitive for id");
		return NULL;
	}

	jsonrpc = json_get_member(buffer, tok, "jsonrpc");
	if (!jsonrpc || jsonrpc->type != JSMN_STRING || !json_tok_streq(buffer, jsonrpc, "2.0")) {
		json_command_malformed(jcon, "null", "jsonrpc: \"2.0\" must be specified in the request");
		return NULL;
	}

	/* Allocate the command off of the `jsonrpc` object and not
	 * the connection since the command may outlive `conn`. */
	c = tal(jcon->ld->jsonrpc, struct command);
	c->jcon = jcon;
	c->send_notifications = jcon->notifications_enabled;
	c->ld = jcon->ld;
	c->pending = false;
	c->json_stream = NULL;
	c->id_is_string = (id->type == JSMN_STRING);
	/* Include "" around string */
	c->id = tal_strndup(c,
			    json_tok_full(buffer, id),
			    json_tok_full_len(id));
	c->mode = CMD_NORMAL;
	c->filter = NULL;
	list_add_tail(&jcon->commands, &c->list);
	tal_add_destructor(c, destroy_command);

	if (!method || !params) {
		return command_fail(c, JSONRPC2_INVALID_REQUEST,
				    method ? "No params" : "No method");
	}

	if (method->type != JSMN_STRING) {
		return command_fail(c, JSONRPC2_INVALID_REQUEST,
				    "Expected string for method");
	}

	c->json_cmd = find_cmd(jcon->ld->jsonrpc, buffer, method);
	if (!c->json_cmd) {
		return command_fail(
		    c, JSONRPC2_METHOD_NOT_FOUND, "Unknown command '%.*s'",
		    method->end - method->start, buffer + method->start);
	}

	if (filter) {
		struct command_result *ret;
		ret = parse_filter(c, "filter", buffer, filter);
		if (ret)
			return ret;
	}

	/* Debug was too chatty, so we use IO here, even though we're
	 * actually just logging the id */
	log_io(jcon->log, LOG_IO_IN, NULL, c->id, NULL, 0);

	if (!command_deprecated_in_ok(c, NULL,
				      c->json_cmd->depr_start,
				      c->json_cmd->depr_end)) {
		return command_fail(c, JSONRPC2_METHOD_NOT_FOUND,
				    "Command %.*s is deprecated",
				    json_tok_full_len(method),
				    json_tok_full(buffer, method));
	}
	if (c->json_cmd->dev_only && !jcon->ld->developer) {
		return command_fail(c, JSONRPC2_METHOD_NOT_FOUND,
				    "Command %.*s is developer-only",
				    json_tok_full_len(method),
				    json_tok_full(buffer, method));
	}

	rpc_hook = tal(c, struct rpc_command_hook_payload);
	rpc_hook->cmd = c;
	/* Duplicate since we might outlive the connection */
	json_dup_contents(rpc_hook, buffer, tok,
			  &rpc_hook->buffer,
			  &rpc_hook->request);

	/* NULL the custom_ values for the hooks */
	rpc_hook->custom_result = NULL;
	rpc_hook->custom_error = NULL;
	rpc_hook->custom_replace = NULL;
	rpc_hook->custom_buffer = NULL;

	*methodname = c->json_cmd->name;
	trace_span_start("lightningd/jsonrpc", &c);
	trace_span_tag(&c, "method", c->json_cmd->name);
	/* They can filter by command name */
	completed = plugin_hook_call_rpc_command(jcon->ld, c->json_cmd->name, c->id, rpc_hook);
	trace_span_end(&c);

	/* If it's deferred, mark it (otherwise, it's completed) */
	if (!completed)
		return command_still_pending(c);
	return NULL;
}

/* Mutual recursion */
static struct io_plan *stream_out_complete(struct io_conn *conn,
					   struct json_stream *js,
					   struct json_connection *jcon);

static struct io_plan *start_json_stream(struct io_conn *conn,
					 struct json_connection *jcon)
{
	struct json_stream *js;

	/* If something has created an output buffer, start streaming. */
	js = list_top(&jcon->jsouts, struct json_stream, list);
	if (js) {
		size_t len;
		const char *p = json_out_contents(js->jout, &len);
		if (len)
			log_io(jcon->log, LOG_IO_OUT, NULL, "", p, len);
		return json_stream_output(js, conn,
					  stream_out_complete, jcon);
	}

	/* Tell reader it can run next command. */
	io_wake(conn);

	/* Once the stop_conn conn is drained, we can shut down. */
	if (jcon->ld->stop_conn == conn && jcon->ld->state == LD_STATE_RUNNING) {
		/* Return us to toplevel lightningd.c */
		log_debug(jcon->ld->log, "io_break: %s", __func__);
		io_break(jcon->ld);
		/* We never come back. */
		return io_out_wait(conn, conn, io_never, conn);
	}

	return io_out_wait(conn, jcon, start_json_stream, jcon);
}

/* Command has completed writing, and we've written it all out to conn. */
static struct io_plan *stream_out_complete(struct io_conn *conn,
					   struct json_stream *js,
					   struct json_connection *jcon)
{
	list_del_from(&jcon->jsouts, &js->list);
	tal_free(js);

	/* Wait for more output. */
	return start_json_stream(conn, jcon);
}

static struct io_plan *read_json(struct io_conn *conn,
				 struct json_connection *jcon)
{
	bool in_transaction = false;
	struct timemono start_time = time_mono();
	size_t len_read;
	const jsmntok_t *toks;
	const char *buffer, *error;
	size_t num_parsed = 0;
	const char *last_method = NULL;
	u64 msec;

	buffer = jsonrpc_newly_read(jcon->json_in, &len_read);
	if (len_read)
		log_io(jcon->log, LOG_IO_IN, NULL, "", buffer, len_read);

	/* We wait for pending output to be consumed, to avoid DoS */
	if (!list_empty(&jcon->jsouts)) {
		return io_wait(conn, conn, read_json, jcon);
	}

again:
	error = jsonrpc_io_parse(tmpctx, jcon->json_in, &toks, &buffer);
	if (error) {
		json_command_malformed(jcon, "null", error);
		if (in_transaction)
			db_commit_transaction(jcon->ld->wallet->db);
		return io_halfclose(conn);
	}

	if (!toks)
		goto read_more;

	if (!in_transaction) {
		db_begin_transaction(jcon->ld->wallet->db);
		in_transaction = true;
	}
	parse_request(jcon, buffer, toks, &last_method);
	jsonrpc_io_parse_done(jcon->json_in);

	/* Don't ever process for more than 100 commands or 250 msec
	 * without giving others a chance */
	msec = time_to_msec(timemono_between(time_mono(), start_time));
	if (num_parsed++ == 100 || msec > 250) {
		static bool printed_once = false;
		db_commit_transaction(jcon->ld->wallet->db);
		log_debug(jcon->log, "Pausing parsing after %zu requests and %"PRIu64"msec (last method=%s)",
			  num_parsed,
			  msec,
			  last_method ? last_method : "NONE");
		if (msec > 5000 && last_method && !printed_once) {
			log_unusual(jcon->log,
				    CI_UNEXPECTED "Request %s took %"PRIu64" milliseconds",
				    last_method, msec);
			printed_once = true;
		}
		/* Call us back, as if we read nothing new */
		return io_always(conn, read_json, jcon);
	}

	if (!jcon->db_batching) {
		db_commit_transaction(jcon->ld->wallet->db);
		in_transaction = false;
	}
	goto again;

read_more:
	if (in_transaction)
		db_commit_transaction(jcon->ld->wallet->db);
	return jsonrpc_io_read(conn, jcon->json_in, read_json, jcon);
}

static struct io_plan *jcon_connected(struct io_conn *conn,
				      struct lightningd *ld)
{
	struct json_connection *jcon;

	/* We live as long as the connection, so we're not a leak. */
	jcon = notleak(tal(conn, struct json_connection));
	jcon->conn = conn;
	jcon->ld = ld;
	list_head_init(&jcon->jsouts);
	jcon->json_in = jsonrpc_io_new(jcon);
	jcon->notifications_enabled = false;
	jcon->db_batching = false;
	jcon->deprecated_ok = ld->deprecated_ok;
	list_head_init(&jcon->commands);

	/* We want to log on destruction, so we free this in destructor. */
	jcon->log = new_logger(ld->log_book, ld->log_book, NULL, "jsonrpc#%i",
			       io_conn_fd(conn));

	tal_add_destructor(jcon, destroy_jcon);

	/* Note that write_json and read_json alternate manually, by waking
	 * each other.  It would be simpler to not use a duplex io, and have
	 * read_json parse one command, then io_wait() for command completion
	 * and go to write_json.
	 *
	 * However, if we ever have notifications, this neat cmd-response
	 * pattern would break down, so we use this trick. */
	return io_duplex(conn,
			 read_json(conn, jcon),
			 start_json_stream(conn, jcon));
}

static struct io_plan *incoming_jcon_connected(struct io_conn *conn,
					       struct lightningd *ld)
{
	/* Lifetime of JSON conn is limited to fd connect time. */
	return jcon_connected(notleak(conn), ld);
}

static void destroy_json_command(struct json_command *command, struct jsonrpc *rpc)
{
	struct cmd_and_usage *cmd;

	if (!strmap_del(&rpc->cmdmap, command->name, &cmd))
		abort();
	tal_free(cmd);
}

static struct cmd_and_usage *command_add(struct jsonrpc *rpc, struct json_command *command)
{
	struct cmd_and_usage *cmd;

	/* Check that we don't clobber a method */
	if (strmap_get(&rpc->cmdmap, command->name))
		return NULL;

	cmd = tal(rpc, struct cmd_and_usage);
	cmd->command = command;
	cmd->usage = NULL;
	strmap_add(&rpc->cmdmap, command->name, cmd);
	return cmd;
}

/* Built-in commands get called to construct usage string via param() */
static void setup_command_usage(struct lightningd *ld,
				struct json_command *command)
{
	const struct command_result *res;
	struct command *dummy;

	/* Call it with minimal cmd, to fill out usagemap */
	dummy = tal(tmpctx, struct command);
	dummy->mode = CMD_USAGE;
	dummy->ld = ld;
	dummy->json_cmd = command;
	res = command->dispatch(dummy, NULL, NULL, NULL);
	assert(res == &param_failed);
	assert(strmap_get(&ld->jsonrpc->cmdmap, command->name)->usage);
}

bool jsonrpc_command_add(struct jsonrpc *rpc, struct json_command *command,
			 const char *usage TAKES)
{
	struct cmd_and_usage *cmd;

	cmd = command_add(rpc, command);
	if (!cmd)
		return false;

	cmd->usage = json_escape_unescape_len(cmd, usage, strlen(usage));
	if (!cmd->usage) {
		tal_free(cmd);
		return false;
	}
	tal_add_destructor2(command, destroy_json_command, rpc);
	return true;
}

static bool jsonrpc_command_add_perm(struct lightningd *ld,
				     struct jsonrpc *rpc,
				     struct json_command *command)
{
	if (!command_add(rpc, command))
		return false;
	setup_command_usage(ld, command);
	return true;
}

static void destroy_jsonrpc(struct jsonrpc *jsonrpc)
{
	strmap_clear(&jsonrpc->cmdmap);
}

static void memleak_help_jsonrpc(struct htable *memtable,
				 struct jsonrpc *jsonrpc)
{
	memleak_scan_strmap(memtable, &jsonrpc->cmdmap);
}

void jsonrpc_setup(struct lightningd *ld)
{
	size_t num_cmdlist;
	struct json_command **commands = get_cmdlist(&num_cmdlist);

	ld->jsonrpc = tal(ld, struct jsonrpc);
	strmap_init(&ld->jsonrpc->cmdmap);
	for (size_t i=0; i<num_cmdlist; i++) {
		if (!jsonrpc_command_add_perm(ld, ld->jsonrpc, commands[i]))
			fatal("Cannot add duplicate command %s",
			      commands[i]->name);
	}
	ld->jsonrpc->rpc_listener = NULL;
	tal_add_destructor(ld->jsonrpc, destroy_jsonrpc);
	memleak_add_helper(ld->jsonrpc, memleak_help_jsonrpc);
}

bool command_usage_only(const struct command *cmd)
{
	return cmd->mode == CMD_USAGE;
}

bool command_dev_apis(const struct command *cmd)
{
	return cmd->ld->developer;
}

void command_log(struct command *cmd, enum log_level level,
		 const char *fmt, ...)
{
	const char *msg;
	va_list ap;

	va_start(ap, fmt);
	msg = tal_vfmt(cmd, fmt, ap);
	log_(cmd->ld->log, level, NULL, level >= LOG_UNUSUAL,
	     "JSON COMMAND %s: %s",
	     cmd->json_cmd->name, msg);
	va_end(ap);
}

void command_set_usage(struct command *cmd, const char *usage TAKES)
{
	struct cmd_and_usage *cmd_and_usage;

	cmd_and_usage = strmap_get(&cmd->ld->jsonrpc->cmdmap, cmd->json_cmd->name);
	assert(!cmd_and_usage->usage);
	cmd_and_usage->usage = tal_strdup(cmd_and_usage, usage);
}

bool command_check_only(const struct command *cmd)
{
	return cmd->mode == CMD_CHECK || cmd->mode == CMD_CHECK_FAILED;
}

void jsonrpc_listen(struct jsonrpc *jsonrpc, struct lightningd *ld)
{
	struct sockaddr_un addr;
	int fd, old_umask, new_umask;
	const char *rpc_filename = ld->rpc_filename;

	/* Should not initialize it twice. */
	assert(!jsonrpc->rpc_listener);

	if (streq(rpc_filename, "/dev/tty")) {
		fd = open(rpc_filename, O_RDWR);
		if (fd == -1)
			err(1, "Opening %s", rpc_filename);
		/* Technically this is a leak, but there's only one */
		notleak(io_new_conn(ld, fd, jcon_connected, ld));
		return;
	}

	fd = socket(AF_UNIX, SOCK_STREAM, 0);
	if (fd < 0) {
		errx(1, "domain socket creation failed");
	}
	if (strlen(rpc_filename) + 1 > sizeof(addr.sun_path))
		errx(1, "rpc filename '%s' too long", rpc_filename);
	strcpy(addr.sun_path, rpc_filename);
	addr.sun_family = AF_UNIX;

	/* Of course, this is racy! */
	if (connect(fd, (struct sockaddr *)&addr, sizeof(addr)) == 0)
		errx(1, "rpc filename '%s' in use", rpc_filename);
	unlink(rpc_filename);

	/* Set the umask according to the desired file mode.  */
	new_umask = ld->rpc_filemode ^ 0777;
	old_umask = umask(new_umask);
	if (bind(fd, (struct sockaddr *)&addr, sizeof(addr)))
		err(1, "Binding rpc socket to '%s'", rpc_filename);
	umask(old_umask);

	if (listen(fd, 128) != 0)
		err(1, "Listening on '%s'", rpc_filename);

	/* All conns will be tal children of jsonrpc: good for freeing later! */
	jsonrpc->rpc_listener
		= io_new_listener(jsonrpc, fd, incoming_jcon_connected, ld);
}

void jsonrpc_stop_listening(struct jsonrpc *jsonrpc)
{
	jsonrpc->rpc_listener = tal_free(jsonrpc->rpc_listener);
}

void jsonrpc_stop_all(struct lightningd *ld)
{
	/* Closes all conns. */
	ld->jsonrpc = tal_free(ld->jsonrpc);
}

static struct command_result *param_command(struct command *cmd,
					    const char *name,
					    const char *buffer,
					    const jsmntok_t *tok,
					    const jsmntok_t **out)
{
	cmd->json_cmd = find_cmd(cmd->jcon->ld->jsonrpc, buffer, tok);
	if (cmd->json_cmd) {
		*out = tok;
		return NULL;
	}

	return command_fail(cmd, JSONRPC2_METHOD_NOT_FOUND,
			    "Unknown command '%.*s'",
			    tok->end - tok->start, buffer + tok->start);
}

struct jsonrpc_notification *jsonrpc_notification_start_noparams(const tal_t *ctx, const char *method)
{
	struct jsonrpc_notification *n = tal(ctx, struct jsonrpc_notification);
	n->method = tal_strdup(n, method);
	n->stream = new_json_stream(n, NULL, NULL);
	json_object_start(n->stream, NULL);
	json_add_string(n->stream, "jsonrpc", "2.0");
	json_add_string(n->stream, "method", method);

	return n;
}

struct jsonrpc_notification *jsonrpc_notification_start(const tal_t *ctx, const char *method)
{
	struct jsonrpc_notification *n = jsonrpc_notification_start_noparams(ctx, method);
	json_object_start(n->stream, "params");

	return n;
}

void jsonrpc_notification_end(struct jsonrpc_notification *n)
{
	json_object_end(n->stream); /* closes '.params' */
	jsonrpc_notification_end_noparams(n);
}

void jsonrpc_notification_end_noparams(struct jsonrpc_notification *n)
{
	json_object_end(n->stream); /* closes '.' */

	/* We guarantee to have \n\n at end of each response. */
	json_stream_append(n->stream, "\n\n", strlen("\n\n"));
}

struct jsonrpc_request *jsonrpc_request_start_(
    const tal_t *ctx, const char *method,
    const char *id_prefix, struct logger *log,
    bool add_header,
    void (*notify_cb)(const char *buffer,
		      const jsmntok_t *methodtok,
		      const jsmntok_t *paramtoks,
		      const jsmntok_t *idtok,
		      void *),
    void (*response_cb)(const char *buffer, const jsmntok_t *toks,
			const jsmntok_t *idtok, void *),
    void *response_cb_arg)
{
	struct jsonrpc_request *r = tal(ctx, struct jsonrpc_request);
	static u64 next_request_id = 0;

	r->id_is_string = true;
	if (id_prefix) {
		/* Strip "" and otherwise sanity-check */
		if (strstarts(id_prefix, "\"")
		    && strlen(id_prefix) > 1
		    && strends(id_prefix, "\"")) {
			id_prefix = tal_strndup(tmpctx, id_prefix + 1,
						strlen(id_prefix) - 2);
		}
		/* We could try escaping, but TBH they're
		 * messing with us at this point! */
		if (json_escape_needed(id_prefix, strlen(id_prefix)))
			id_prefix = "weird-id";

		r->id = tal_fmt(r, "\"%s/cln:%s#%"PRIu64"\"",
				id_prefix, method, next_request_id);
	} else {
		r->id = tal_fmt(r, "\"cln:%s#%"PRIu64"\"", method, next_request_id);
	}
	if (taken(id_prefix))
		tal_free(id_prefix);
	next_request_id++;
	r->notify_cb = notify_cb;
	r->response_cb = response_cb;
	r->response_cb_arg = response_cb_arg;
	r->method = tal_strdup(r, method);
	r->stream = new_json_stream(r, NULL, log);

	/* Disabling this serves as an escape hatch for plugin code to
	 * get a raw request to paste into, but get a valid request-id
	 * assigned. */
	if (add_header) {
		json_object_start(r->stream, NULL);
		json_add_string(r->stream, "jsonrpc", "2.0");
		json_add_id(r->stream, r->id);
		json_add_string(r->stream, "method", method);
		json_object_start(r->stream, "params");
	}
	if (log)
		log_io(log, LOG_IO_OUT, NULL, r->id, NULL, 0);

	return r;
}

void jsonrpc_request_end(struct jsonrpc_request *r)
{
	json_object_end(r->stream); /* closes '.params' */
	json_object_end(r->stream); /* closes '.' */

	/* We guarantee to have \n\n at end of each response. */
	json_stream_append(r->stream, "\n\n", strlen("\n\n"));
}

static struct command_result *json_check(struct command *cmd,
					 const char *buffer,
					 const jsmntok_t *obj,
					 const jsmntok_t *params)
{
	jsmntok_t *mod_params;
	const jsmntok_t *name_tok;
	struct command_result *res;
	struct lightningd *ld = cmd->ld;

	if (cmd->mode == CMD_USAGE) {
		mod_params = NULL;
	} else {
		mod_params = json_tok_copy(cmd, params);
	}

	/* Replaces cmd->json_cmd: */
	if (!param(cmd, buffer, mod_params,
		   p_req("command_to_check", param_command, &name_tok),
		   p_opt_any(),
		   NULL))
		return command_param_failed();

	cmd->mode = CMD_CHECK;
	/* Make *sure* it doesn't try to manip db! */
	db_set_readonly(ld->wallet->db, true);

	/* Raw check hook is needed for plugins */
	if (cmd->json_cmd->check) {
		res = cmd->json_cmd->check(cmd, buffer, obj, params);
	} else {
		/* Point name_tok to the name, not the value */
		if (params->type == JSMN_OBJECT)
			name_tok--;

		json_tok_remove(&mod_params, mod_params, name_tok, 1);

		res = cmd->json_cmd->dispatch(cmd, buffer, mod_params, mod_params);
	}
	db_set_readonly(ld->wallet->db, false);

	return res;
}

static const struct json_command check_command = {
	"check",
	json_check,
};
AUTODATA(json_command, &check_command);

static struct command_result *json_notifications(struct command *cmd,
						 const char *buffer,
						 const jsmntok_t *obj UNNEEDED,
						 const jsmntok_t *params)
{
	bool *enable;

	if (!param(cmd, buffer, params,
		   p_req("enable", param_bool, &enable),
		   NULL))
		return command_param_failed();

	/* Catch the case where they sent this command then hung up. */
	if (cmd->jcon)
		cmd->jcon->notifications_enabled = *enable;
	return command_success(cmd, json_stream_success(cmd));
}

static const struct json_command notifications_command = {
	"notifications",
	json_notifications,
};
AUTODATA(json_command, &notifications_command);

static struct command_result *json_batching(struct command *cmd,
					    const char *buffer,
					    const jsmntok_t *obj UNNEEDED,
					    const jsmntok_t *params)
{
	bool *enable;

	if (!param(cmd, buffer, params,
		   p_req("enable", param_bool, &enable),
		   NULL))
		return command_param_failed();

	/* Catch the case where they sent this command then hung up. */
	if (cmd->jcon)
		cmd->jcon->db_batching = *enable;
	return command_success(cmd, json_stream_success(cmd));
}

static const struct json_command batching_command = {
	"batching",
	json_batching,
};
AUTODATA(json_command, &batching_command);

static struct command_result *json_deprecations(struct command *cmd,
						const char *buffer,
						const jsmntok_t *obj UNNEEDED,
						const jsmntok_t *params)
{
	bool *enable;

	if (!param(cmd, buffer, params,
		   p_req("enable", param_bool, &enable),
		   NULL))
		return command_param_failed();

	/* Catch the case where they sent this command then hung up. */
	if (cmd->jcon)
		cmd->jcon->deprecated_ok = *enable;
	return command_success(cmd, json_stream_success(cmd));
}

static const struct json_command deprecations_command = {
	"deprecations",
	json_deprecations,
};
AUTODATA(json_command, &deprecations_command);