use lightning::util::logger::Logger;
use lightning::util::ser::Writeable;
use tokio::sync::mpsc;
-use tokio_postgres::NoTls;
use crate::config;
use crate::types::GossipMessage;
}
pub(crate) async fn persist_gossip(&mut self) {
- let connection_config = config::db_connection_config();
- let (mut client, connection) =
- connection_config.connect(NoTls).await.unwrap();
-
- tokio::spawn(async move {
- if let Err(e) = connection.await {
- panic!("connection error: {}", e);
- }
- });
+ let mut client = crate::connect_to_db().await;
{
// initialize the database
config::upgrade_db(cur_schema[0].get(0), &mut client).await;
}
+ let preparation = client.execute("set time zone UTC", &[]).await;
+ if let Err(preparation_error) = preparation {
+ panic!("db preparation error: {}", preparation_error);
+ }
+
let initialization = client
.execute(
// TODO: figure out a way to fix the id value without Postgres complaining about
}
match &gossip_message {
- GossipMessage::ChannelAnnouncement(announcement) => {
+ GossipMessage::ChannelAnnouncement(announcement, _) => {
let scid = announcement.contents.short_channel_id as i64;
// start with the type prefix, which is already known a priori
&announcement_signed
])).await.unwrap().unwrap();
}
- GossipMessage::ChannelUpdate(update) => {
+ GossipMessage::ChannelUpdate(update, seen_override) => {
let scid = update.contents.short_channel_id as i64;
let timestamp = update.contents.timestamp as i64;
let mut update_signed = Vec::new();
update.write(&mut update_signed).unwrap();
- tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
- .execute("INSERT INTO channel_updates (\
+ let insertion_statement = if cfg!(test) {
+ "INSERT INTO channel_updates (\
short_channel_id, \
timestamp, \
+ seen, \
channel_flags, \
direction, \
disable, \
fee_proportional_millionths, \
htlc_maximum_msat, \
blob_signed \
- ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ON CONFLICT DO NOTHING", &[
+ ) VALUES ($1, $2, TO_TIMESTAMP($3), $4, $5, $6, $7, $8, $9, $10, $11, $12) ON CONFLICT DO NOTHING"
+ } else {
+ "INSERT INTO channel_updates (\
+ short_channel_id, \
+ timestamp, \
+ channel_flags, \
+ direction, \
+ disable, \
+ cltv_expiry_delta, \
+ htlc_minimum_msat, \
+ fee_base_msat, \
+ fee_proportional_millionths, \
+ htlc_maximum_msat, \
+ blob_signed \
+ ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ON CONFLICT DO NOTHING"
+ };
+
+ // this may not be used outside test cfg
+ let _seen_timestamp = seen_override.unwrap_or(timestamp as u32) as f64;
+
+ tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
+ .execute(insertion_statement, &[
&scid,
×tamp,
+ #[cfg(test)]
+ &_seen_timestamp,
&(update.contents.flags as i16),
&direction,
&disable,