bkpr: track channel rebalances, display in listincome

Track rebalances, and report income events for them.

Previously `listincome` would report:
	- invoice event, debit, outgoing channel
	- invoice_fee event, debit, outgoing channel
	- invoice event, credit, inbound channel

Now reports:
	- rebalance_fee, debit, outgoing channel
	(same value as invoice_fee above)

Note: only applies on channel events; if a rebalance falls to chain
we'll use the older style of accounting.

Changelog-None

doc/lightning-bkpr-listaccountevents.7.md

@@ -45,6 +45,7 @@ If **type** is "onchain_fee":
 
 If **type** is "channel":
   - **fees_msat** (msat, optional): Amount paid in fees
+  - **is_rebalance** (boolean, optional): Is this payment part of a rebalance
   - **payment_id** (hex, optional): lightning payment identifier. For an htlc, this will be the preimage.
   - **part_id** (u32, optional): Counter for multi-part payments
 
@@ -66,4 +67,4 @@ RESOURCES
 
 Main web site: <https://github.com/ElementsProject/lightning>
 
-[comment]: # ( SHA256STAMP:f8538b1d1e6cda7cd801690e5c09741c8a843b27cc922065598914516c16d2b3)
+[comment]: # ( SHA256STAMP:8568188808cb649d7182ffb628950b93b18406a0498b5b6768371bc94375e258)

doc/schemas/bkpr-listaccountevents.schema.json

@@ -165,6 +165,10 @@
                   "type": "msat",
                   "description": "Amount paid in fees"
                 },
+                "is_rebalance": {
+                  "type": "boolean",
+                  "description": "Is this payment part of a rebalance"
+                },
                 "payment_id": {
                   "type": "hex",
                   "description": "lightning payment identifier. For an htlc, this will be the preimage."

plugins/bkpr/account_entry.c

@@ -8,6 +8,7 @@ static const char *tags[] = {
 	"journal_entry",
 	"penalty_adj",
 	"invoice_fee",
+	"rebalance_fee",
 };
 
 const char *account_entry_tag_str(enum account_entry_tag tag)

plugins/bkpr/account_entry.h

@@ -2,11 +2,12 @@
 #define LIGHTNING_PLUGINS_BKPR_ACCOUNT_ENTRY_H
 #include "config.h"
 
-#define NUM_ACCOUNT_ENTRY_TAGS (INVOICEFEE + 1)
+#define NUM_ACCOUNT_ENTRY_TAGS (REBALANCEFEE + 1)
 enum account_entry_tag {
 	JOURNAL_ENTRY = 0,
 	PENALTY_ADJ = 1,
 	INVOICEFEE = 2,
+	REBALANCEFEE= 3,
 };
 
 /* Convert an enum into a string */

plugins/bkpr/bookkeeper.c

@@ -1645,6 +1645,7 @@ parse_and_log_channel_move(struct command *cmd,
 	e->timestamp = timestamp;
 	e->tag = mvt_tag_str(tags[0]);
 	e->desc = tal_steal(e, desc);
+	e->rebalance_id = NULL;
 
 	/* Go find the account for this event */
 	db_begin_transaction(db);
@@ -1656,21 +1657,27 @@ parse_and_log_channel_move(struct command *cmd,
 			   acct_name);
 
 	log_channel_event(db, acct, e);
-	db_commit_transaction(db);
 
 	/* Check for invoice desc data, necessary */
 	if (e->payment_id) {
 		for (size_t i = 0; i < tal_count(tags); i++) {
 			if (tags[i] != INVOICE)
 				continue;
 
+			/* We only do rebalance checks for debits,
+			 * the credit event always arrives first */
+			if (!amount_msat_zero(e->debit))
+				maybe_record_rebalance(db, e);
+
+			db_commit_transaction(db);
 			/* Keep memleak happy */
 			tal_steal(tmpctx, e);
 			return lookup_invoice_desc(cmd, e->credit,
 						   e->payment_id);
 		}
 	}
 
+	db_commit_transaction(db);
 	return notification_handled(cmd);
 }
 

plugins/bkpr/chain_event.h

@@ -34,6 +34,9 @@ struct chain_event {
 	 * we'll need to watch it for longer */
 	bool stealable;
 
+	/* Is this a rebalance event? */
+	bool rebalance;
+
 	/* Amount we received in this event */
 	struct amount_msat credit;
 

plugins/bkpr/channel_event.c

@@ -27,6 +27,7 @@ struct channel_event *new_channel_event(const tal_t *ctx,
 	ev->part_id = part_id;
 	ev->timestamp = timestamp;
 	ev->desc = NULL;
+	ev->rebalance_id = NULL;
 
 	return ev;
 }
@@ -50,5 +51,6 @@ void json_add_channel_event(struct json_stream *out,
 	json_add_u64(out, "timestamp", ev->timestamp);
 	if (ev->desc)
 		json_add_string(out, "description", ev->desc);
+	json_add_bool(out, "is_rebalance", ev->rebalance_id != NULL);
 	json_object_end(out);
 }

plugins/bkpr/channel_event.h

@@ -46,6 +46,9 @@ struct channel_event {
 
 	/* Description, usually from invoice */
 	const char *desc;
+
+	/* ID of paired event, iff is a rebalance */
+	u64 *rebalance_id;
 };
 
 struct channel_event *new_channel_event(const tal_t *ctx,

plugins/bkpr/db.c

@@ -98,6 +98,7 @@ static struct migration db_migrations[] = {
 	{SQL("ALTER TABLE chain_events ADD stealable INTEGER;"), NULL},
 	{SQL("ALTER TABLE chain_events ADD ev_desc TEXT DEFAULT NULL;"), NULL},
 	{SQL("ALTER TABLE channel_events ADD ev_desc TEXT DEFAULT NULL;"), NULL},
+	{SQL("ALTER TABLE channel_events ADD rebalance_id BIGINT DEFAULT NULL;"), NULL},
 };
 
 static bool db_migrate(struct plugin *p, struct db *db, bool *created)

plugins/bkpr/incomestmt.c

@@ -222,6 +222,16 @@ static struct income_event *paid_invoice_fee(const tal_t *ctx,
 	return iev;
 }
 
+static struct income_event *rebalance_fee(const tal_t *ctx,
+					  struct channel_event *ev)
+{
+	struct income_event *iev;
+	iev = channel_to_income(ctx, ev, AMOUNT_MSAT(0), ev->fees);
+	iev->tag = tal_free(ev->tag);
+	iev->tag = (char *)account_entry_tag_str(REBALANCEFEE);
+	return iev;
+}
+
 static struct income_event *maybe_channel_income(const tal_t *ctx,
 						 struct channel_event *ev)
 {
@@ -235,6 +245,10 @@ static struct income_event *maybe_channel_income(const tal_t *ctx,
 	}
 
 	if (streq(ev->tag, "invoice")) {
+		/* Skip events for rebalances */
+		if (ev->rebalance_id)
+			return NULL;
+
 		/* If it's a payment, we note fees separately */
 		if (!amount_msat_zero(ev->debit)) {
 			struct amount_msat paid;
@@ -383,11 +397,14 @@ struct income_event **list_income_events(const tal_t *ctx,
 			if (ev)
 				tal_arr_expand(&evs, ev);
 
-			/* Breakout fees on sent payments, if present */
+			/* Report fees on payments, if present */
 			if (streq(chan->tag, "invoice")
 			    && !amount_msat_zero(chan->debit)
 			    && !amount_msat_zero(chan->fees)) {
-				ev = paid_invoice_fee(evs, chan);
+				if (!chan->rebalance_id)
+					ev = paid_invoice_fee(evs, chan);
+				else
+					ev = rebalance_fee(evs, chan);
 				tal_arr_expand(&evs, ev);
 			}
 

plugins/bkpr/recorder.c

@@ -112,9 +112,29 @@ static struct channel_event *stmt2channel_event(const tal_t *ctx, struct db_stmt
 	else
 		e->desc = NULL;
 
+	if (!db_col_is_null(stmt, "e.rebalance_id")) {
+		e->rebalance_id = tal(e, u64);
+		*e->rebalance_id = db_col_u64(stmt, "e.rebalance_id");
+	} else
+		e->rebalance_id = NULL;
+
 	return e;
 }
 
+static struct rebalance *stmt2rebalance(const tal_t *ctx, struct db_stmt *stmt)
+{
+	struct rebalance *r = tal(ctx, struct rebalance);
+
+	r->in_ev_id = db_col_u64(stmt, "in_e.id");
+	r->out_ev_id = db_col_u64(stmt, "out_e.id");
+	r->in_acct_name = db_col_strdup(r, stmt, "in_acct.name");
+	r->out_acct_name = db_col_strdup(r, stmt, "out_acct.name");
+	db_col_amount_msat(stmt, "in_e.credit", &r->rebal_msat);
+	db_col_amount_msat(stmt, "out_e.fees", &r->fee_msat);
+
+	return r;
+}
+
 struct chain_event **list_chain_events_timebox(const tal_t *ctx,
 					       struct db *db,
 					       u64 start_time,
@@ -889,6 +909,7 @@ struct channel_event **list_channel_events_timebox(const tal_t *ctx,
 				     ", e.part_id"
 				     ", e.timestamp"
 				     ", e.ev_desc"
+				     ", e.rebalance_id"
 				     " FROM channel_events e"
 				     " LEFT OUTER JOIN accounts a"
 				     " ON a.id = e.account_id"
@@ -936,6 +957,7 @@ struct channel_event **account_get_channel_events(const tal_t *ctx,
 				     ", e.part_id"
 				     ", e.timestamp"
 				     ", e.ev_desc"
+				     ", e.rebalance_id"
 				     " FROM channel_events e"
 				     " LEFT OUTER JOIN accounts a"
 				     " ON a.id = e.account_id"
@@ -1339,9 +1361,10 @@ void log_channel_event(struct db *db,
 				     ", part_id"
 				     ", timestamp"
 				     ", ev_desc"
+				     ", rebalance_id"
 				     ")"
 				     " VALUES"
-				     " (?, ?, ?, ?, ?, ?, ?, ?, ?, ?);"));
+				     " (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);"));
 
 	db_bind_u64(stmt, 0, acct->db_id);
 	db_bind_text(stmt, 1, e->tag);
@@ -1360,6 +1383,11 @@ void log_channel_event(struct db *db,
 	else
 		db_bind_null(stmt, 9);
 
+	if (e->rebalance_id)
+		db_bind_u64(stmt, 10, *e->rebalance_id);
+	else
+		db_bind_null(stmt, 10);
+
 	db_exec_prepared_v2(stmt);
 	e->db_id = db_last_insert_id_v2(stmt);
 	e->acct_db_id = acct->db_id;
@@ -1644,6 +1672,93 @@ static char *is_closed_channel_txid(const tal_t *ctx, struct db *db,
 	return NULL;
 }
 
+void maybe_record_rebalance(struct db *db,
+			    struct channel_event *out)
+{
+	/* If there's a matching credit event, this is
+	 * a rebalance. Mark everything with the payment_id
+	 * and amt as such. If you repeat a payment_id
+	 * with the same amt, they'll be marked as rebalances
+	 * also */
+	struct db_stmt *stmt;
+	struct amount_msat credit;
+	bool ok;
+
+	/* The amount of we were credited is debit - fees */
+	ok = amount_msat_sub(&credit, out->debit, out->fees);
+	assert(ok);
+
+	stmt = db_prepare_v2(db, SQL("SELECT "
+				     "  e.id"
+				     " FROM channel_events e"
+				     " WHERE e.payment_id = ?"
+				     " AND e.credit = ?"
+				     " AND e.rebalance_id IS NULL"));
+
+	db_bind_sha256(stmt, 0, out->payment_id);
+	db_bind_amount_msat(stmt, 1, &credit);
+	db_query_prepared(stmt);
+
+	if (!db_step(stmt)) {
+		/* No matching invoice found */
+		tal_free(stmt);
+		return;
+	}
+
+	/* We just take the first one */
+	out->rebalance_id = tal(out, u64);
+	*out->rebalance_id = db_col_u64(stmt, "e.id");
+	tal_free(stmt);
+
+	/* Set rebalance flag on both records */
+	stmt = db_prepare_v2(db, SQL("UPDATE channel_events SET"
+				     "  rebalance_id = ?"
+				     " WHERE"
+				     " id = ?"));
+	db_bind_u64(stmt, 0, *out->rebalance_id);
+	db_bind_u64(stmt, 1, out->db_id);
+	db_exec_prepared_v2(take(stmt));
+
+	stmt = db_prepare_v2(db, SQL("UPDATE channel_events SET"
+				     "  rebalance_id = ?"
+				     " WHERE"
+				     " id = ?"));
+	db_bind_u64(stmt, 0, out->db_id);
+	db_bind_u64(stmt, 1, *out->rebalance_id);
+	db_exec_prepared_v2(take(stmt));
+}
+
+struct rebalance **list_rebalances(const tal_t *ctx, struct db *db)
+{
+	struct rebalance **result;
+	struct db_stmt *stmt;
+
+	stmt = db_prepare_v2(db, SQL("SELECT "
+				     "  in_e.id"
+				     ", out_e.id"
+				     ", in_acct.name"
+				     ", out_acct.name"
+				     ", in_e.credit"
+				     ", out_e.fees"
+				     " FROM channel_events in_e"
+				     " LEFT OUTER JOIN channel_events out_e"
+				     " ON in_e.rebalance_id = out_e.id"
+				     " LEFT OUTER JOIN accounts out_acct"
+				     " ON out_acct.id = out_e.account_id"
+				     " LEFT OUTER JOIN accounts in_acct"
+				     " ON in_acct.id = in_e.account_id"
+				     " WHERE in_e.rebalance_id IS NOT NULL"
+				     "  AND in_e.credit > 0"));
+	db_query_prepared(stmt);
+	result = tal_arr(ctx, struct rebalance *, 0);
+	while (db_step(stmt)) {
+		struct rebalance *r = stmt2rebalance(result, stmt);
+		tal_arr_expand(&result, r);
+	}
+	tal_free(stmt);
+	return result;
+}
+
 char *maybe_update_onchain_fees(const tal_t *ctx, struct db *db,
 			        struct bitcoin_txid *txid)
 {

plugins/bkpr/recorder.h

@@ -41,6 +41,15 @@ struct txo_set {
 	struct txo_pair **pairs;
 };
 
+struct rebalance {
+	u64 in_ev_id;
+	u64 out_ev_id;
+	char *in_acct_name;
+	char *out_acct_name;
+	struct amount_msat rebal_msat;
+	struct amount_msat fee_msat;
+};
+
 /* Get all accounts */
 struct account **list_accounts(const tal_t *ctx, struct db *db);
 
@@ -200,6 +209,14 @@ void add_payment_hash_desc(struct db *db,
  * height an input was spent into */
 void maybe_closeout_external_deposits(struct db *db, struct chain_event *ev);
 
+/* Keep track of rebalancing payments (payments paid to/from ourselves.
+ * Returns true if was rebalance */
+void maybe_record_rebalance(struct db *db,
+			    struct channel_event *out);
+
+/* List all rebalances */
+struct rebalance **list_rebalances(const tal_t *ctx, struct db *db);
+
 /* Log a channel event */
 void log_channel_event(struct db *db,
 		       const struct account *acct,