- let reminder_threshold_timestamp = SystemTime::now().checked_sub(config::CHANNEL_REMINDER_AGE).unwrap();
-
- let params: [&(dyn tokio_postgres::types::ToSql + Sync); 2] =
- [&channel_ids, &reminder_threshold_timestamp];
- let older_latest_directional_updates = client.query_raw("
- SELECT short_channel_id FROM (
- SELECT DISTINCT ON (short_channel_id) *
- FROM (
- SELECT DISTINCT ON (short_channel_id, direction) short_channel_id, seen
- FROM channel_updates
- WHERE short_channel_id = any($1)
- ORDER BY short_channel_id ASC, direction ASC, seen DESC
- ) AS directional_last_seens
- ORDER BY short_channel_id ASC, seen ASC
- ) AS distinct_chans
- WHERE distinct_chans.seen <= $2
- ", params).await.unwrap();
- let mut pinned_updates = Box::pin(older_latest_directional_updates);
-
+ let reminder_threshold_timestamp = current_time.checked_sub(config::CHANNEL_REMINDER_AGE).unwrap().duration_since(UNIX_EPOCH).unwrap().as_secs() as f64;
+
+ log_info!(logger, "Fetch first time we saw the current value combination for each direction (prior mutations excepted)");
+ let reminder_lookup_threshold_timestamp = current_time.checked_sub(config::CHANNEL_REMINDER_AGE * 3).unwrap().duration_since(UNIX_EPOCH).unwrap().as_secs() as f64;
+ let params: [&(dyn tokio_postgres::types::ToSql + Sync); 2] = [&channel_ids, &reminder_lookup_threshold_timestamp];
+
+ /*
+ What exactly is the below query doing?
+
+ First, the inner query groups all channel updates by their scid/direction combination,
+ and then sorts those in reverse chronological order by the "seen" column.
+
+ Then, each row is annotated based on whether its subsequent row for the same scid/direction
+ combination has a different value for any one of these six fields:
+ disable, cltv_expiry_delta, htlc_minimum_msat, fee_base_msat, fee_proportional_millionths, htlc_maximum_msat
+ Those are simply the properties we use to keep track of channel mutations.
+
+ The outer query takes all of those results and selects the first value that has a distinct
+ successor for each scid/direction combination. That yields the first instance at which
+ a given channel configuration was received after any prior mutations.
+
+ Knowing that, we can check whether or not there have been any mutations within the
+ reminder requirement window. Because we only care about that window (and potentially the
+ 2-week-window), we pre-filter the scanned updates by only those that were received within
+ 3x the timeframe that we consider necessitates reminders.
+ */
+
+ let mutated_updates = client.query_raw("
+ SELECT DISTINCT ON (short_channel_id, direction) short_channel_id, direction, blob_signed, CAST(EXTRACT('epoch' from seen) AS BIGINT) AS seen FROM (
+ SELECT short_channel_id, direction, timestamp, seen, blob_signed, COALESCE (
+ disable<>lead(disable) OVER w1
+ OR
+ cltv_expiry_delta<>lead(cltv_expiry_delta) OVER w1
+ OR
+ htlc_minimum_msat<>lead(htlc_minimum_msat) OVER w1
+ OR
+ fee_base_msat<>lead(fee_base_msat) OVER w1
+ OR
+ fee_proportional_millionths<>lead(fee_proportional_millionths) OVER w1
+ OR
+ htlc_maximum_msat<>lead(htlc_maximum_msat) OVER w1,
+ TRUE
+ ) has_distinct_successor
+ FROM channel_updates
+ WHERE short_channel_id = any($1) AND seen >= TO_TIMESTAMP($2)
+ WINDOW w1 AS (PARTITION BY short_channel_id, direction ORDER BY seen DESC)
+ ) _
+ WHERE has_distinct_successor
+ ORDER BY short_channel_id ASC, direction ASC, timestamp DESC
+ ", params).await.unwrap();
+
+ let mut pinned_updates = Box::pin(mutated_updates);
+ let mut older_latest_directional_update_count = 0;