Make pindexer more easily embeddable as a library (#4712)

## Describe your changes

We don't expose cometindex as a library, but we do pindexer, and this
adds a few tweaks to make pindexer possible to run inside an external
piece of Rust code. This is useful, because it allows people to colocate
their indexer and their block explorer, for example.

## Checklist before requesting a review

- [x] If this code contains consensus-breaking changes, I have added the
"consensus-breaking" label. Otherwise, I declare my belief that there
are not consensus-breaking changes, for the following reason:

  > Doesn't touch the core protocol.

crates/bin/pindexer/Cargo.toml

@@ -11,15 +11,16 @@ publish = false
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+anyhow = {workspace = true}
+clap = {workspace = true}
+cometindex = {workspace = true}
+penumbra-shielded-pool = {workspace = true, default-features = false}
+penumbra-stake = {workspace = true, default-features = false}
+penumbra-app = {workspace = true, default-features = false}
+penumbra-num = {workspace = true, default-features = false}
+penumbra-asset = {workspace = true, default-features = false}
+penumbra-proto = {workspace = true, default-features = false}
+tokio = {workspace = true, features = ["full"]}
+serde_json = {workspace = true}
 sqlx = { workspace = true, features = ["chrono"] }
-cometindex = { workspace = true }
-penumbra-shielded-pool = { workspace = true, default-features = false }
-penumbra-stake = { workspace = true, default-features = false }
-penumbra-app = { workspace = true, default-features = false }
-penumbra-num = { workspace = true, default-features = false }
-penumbra-asset = { workspace = true, default-features = false }
-penumbra-proto = { workspace = true, default-features = false }
-tokio = { workspace = true, features = ["full"] }
-anyhow = { workspace = true }
-serde_json = { workspace = true }
-tracing = { workspace = true }
+tracing = {workspace = true}

crates/bin/pindexer/src/block.rs

@@ -1,6 +1,6 @@
 use cometindex::{async_trait, sqlx, AppView, ContextualizedEvent, PgTransaction};
 use penumbra_proto::{core::component::sct::v1 as pb, event::ProtoEvent};
-use sqlx::types::chrono::DateTime;
+use sqlx::{types::chrono::DateTime, PgPool};
 
 #[derive(Debug)]
 pub struct Block {}
@@ -36,6 +36,7 @@ CREATE TABLE IF NOT EXISTS block_details (
         &self,
         dbtx: &mut PgTransaction,
         event: &ContextualizedEvent,
+        _src_db: &PgPool,
     ) -> Result<(), anyhow::Error> {
         let pe = pb::EventBlockRoot::from_event(event.as_ref())?;
         let timestamp = pe.timestamp.expect("Block has no timestamp");

crates/bin/pindexer/src/lib.rs

@@ -1,4 +1,4 @@
-pub use cometindex::{AppView, Indexer};
+pub use cometindex::{opt::Options, AppView, ContextualizedEvent, Indexer, PgPool, PgTransaction};
 
 mod indexer_ext;
 pub use indexer_ext::IndexerExt;

crates/bin/pindexer/src/main.rs

@@ -1,10 +1,11 @@
 use anyhow::Result;
+use clap::Parser as _;
 use pindexer::block::Block;
-use pindexer::{Indexer, IndexerExt as _};
+use pindexer::{Indexer, IndexerExt as _, Options};
 
 #[tokio::main]
 async fn main() -> Result<()> {
-    Indexer::new()
+    Indexer::new(Options::parse())
         .with_default_tracing()
         .with_default_penumbra_app_views()
         .with_index(Block {})

crates/bin/pindexer/src/shielded_pool/fmd.rs

@@ -1,4 +1,4 @@
-use cometindex::{async_trait, sqlx, AppView, ContextualizedEvent, PgTransaction};
+use cometindex::{async_trait, sqlx, AppView, ContextualizedEvent, PgPool, PgTransaction};
 use penumbra_proto::{core::component::shielded_pool::v1 as pb, event::ProtoEvent};
 
 #[derive(Debug)]
@@ -34,6 +34,7 @@ CREATE TABLE IF NOT EXISTS shielded_pool_fmd_clue_set (
         &self,
         dbtx: &mut PgTransaction,
         event: &ContextualizedEvent,
+        _src_db: &PgPool,
     ) -> Result<(), anyhow::Error> {
         let pe = pb::EventBroadcastClue::from_event(event.as_ref())?;
 

crates/bin/pindexer/src/stake/delegation_txs.rs

@@ -1,5 +1,5 @@
 use anyhow::{anyhow, Result};
-use cometindex::{async_trait, sqlx, AppView, ContextualizedEvent, PgTransaction};
+use cometindex::{async_trait, sqlx, AppView, ContextualizedEvent, PgPool, PgTransaction};
 use penumbra_num::Amount;
 use penumbra_proto::{core::component::stake::v1 as pb, event::ProtoEvent};
 
@@ -54,6 +54,7 @@ impl AppView for DelegationTxs {
         &self,
         dbtx: &mut PgTransaction,
         event: &ContextualizedEvent,
+        _src_db: &PgPool,
     ) -> Result<()> {
         let pe = pb::EventDelegate::from_event(event.as_ref())?;
 

crates/bin/pindexer/src/stake/missed_blocks.rs

@@ -1,5 +1,5 @@
 use anyhow::Result;
-use cometindex::{async_trait, sqlx, AppView, ContextualizedEvent, PgTransaction};
+use cometindex::{async_trait, sqlx, AppView, ContextualizedEvent, PgPool, PgTransaction};
 
 use penumbra_proto::{core::component::stake::v1 as pb, event::ProtoEvent};
 
@@ -52,6 +52,7 @@ impl AppView for MissedBlocks {
         &self,
         dbtx: &mut PgTransaction,
         event: &ContextualizedEvent,
+        _src_db: &PgPool,
     ) -> Result<(), anyhow::Error> {
         let pe = pb::EventValidatorMissedBlock::from_event(event.as_ref())?;
         let ik_bytes = pe

crates/bin/pindexer/src/stake/slashings.rs

@@ -1,5 +1,5 @@
 use anyhow::{anyhow, Result};
-use cometindex::{async_trait, sqlx, AppView, ContextualizedEvent, PgTransaction};
+use cometindex::{async_trait, sqlx, AppView, ContextualizedEvent, PgPool, PgTransaction};
 
 use penumbra_proto::{core::component::stake::v1 as pb, event::ProtoEvent};
 use penumbra_stake::IdentityKey;
@@ -52,6 +52,7 @@ impl AppView for Slashings {
         &self,
         dbtx: &mut PgTransaction,
         event: &ContextualizedEvent,
+        _src_db: &PgPool,
     ) -> Result<(), anyhow::Error> {
         let pe = pb::EventSlashingPenaltyApplied::from_event(event.as_ref())?;
         let ik = IdentityKey::try_from(

crates/bin/pindexer/src/stake/undelegation_txs.rs

@@ -1,5 +1,5 @@
 use anyhow::{anyhow, Result};
-use cometindex::{async_trait, sqlx, AppView, ContextualizedEvent, PgTransaction};
+use cometindex::{async_trait, sqlx, AppView, ContextualizedEvent, PgPool, PgTransaction};
 use penumbra_num::Amount;
 use penumbra_proto::{core::component::stake::v1 as pb, event::ProtoEvent};
 
@@ -54,6 +54,7 @@ impl AppView for UndelegationTxs {
         &self,
         dbtx: &mut PgTransaction,
         event: &ContextualizedEvent,
+        _src_db: &PgPool,
     ) -> Result<()> {
         let pe = pb::EventUndelegate::from_event(event.as_ref())?;
 

crates/bin/pindexer/src/stake/validator_set.rs

@@ -1,7 +1,7 @@
 use std::collections::BTreeMap;
 
 use anyhow::{anyhow, Context, Result};
-use cometindex::{async_trait, sqlx, AppView, ContextualizedEvent, PgTransaction};
+use cometindex::{async_trait, sqlx, AppView, ContextualizedEvent, PgPool, PgTransaction};
 
 use penumbra_app::genesis::AppState;
 use penumbra_asset::asset;
@@ -68,6 +68,7 @@ impl AppView for ValidatorSet {
         &self,
         dbtx: &mut PgTransaction,
         event: &ContextualizedEvent,
+        _src_db: &PgPool,
     ) -> Result<(), anyhow::Error> {
         match event.event.kind.as_str() {
             "penumbra.core.component.stake.v1.EventValidatorDefinitionUpload" => {

crates/util/cometindex/examples/fmd_clues.rs

@@ -1,5 +1,7 @@
 use anyhow::Result;
-use cometindex::{async_trait, AppView, ContextualizedEvent, Indexer, PgTransaction};
+use clap::Parser;
+use cometindex::{async_trait, opt::Options, AppView, ContextualizedEvent, Indexer, PgTransaction};
+use sqlx::PgPool;
 
 // This example is silly because it doesn't do any "compilation" of the raw
 // events, so it's only useful as an example of exercising the harness and the
@@ -39,6 +41,7 @@ CREATE TABLE IF NOT EXISTS fmd_clues_example (
         &self,
         dbtx: &mut PgTransaction,
         event: &ContextualizedEvent,
+        _src_db: &PgPool,
     ) -> Result<(), anyhow::Error> {
         // this is just an example in the integration tests, so we don't want to do any
         // - queries against existing table state
@@ -72,7 +75,7 @@ CREATE TABLE IF NOT EXISTS fmd_clues_example (
 
 #[tokio::main]
 async fn main() -> Result<()> {
-    Indexer::new()
+    Indexer::new(Options::parse())
         .with_default_tracing()
         // add as many indexers as you want
         .with_index(FmdCluesExample {})

crates/util/cometindex/src/index.rs

@@ -1,4 +1,5 @@
 use async_trait::async_trait;
+pub use sqlx::PgPool;
 use sqlx::{Postgres, Transaction};
 
 use crate::ContextualizedEvent;
@@ -20,5 +21,6 @@ pub trait AppView: std::fmt::Debug {
         &self,
         dbtx: &mut PgTransaction,
         event: &ContextualizedEvent,
+        src_db: &PgPool,
     ) -> Result<(), anyhow::Error>;
 }

crates/util/cometindex/src/indexer.rs

@@ -1,9 +1,8 @@
 use std::pin::Pin;
 
 use anyhow::{Context as _, Result};
-use clap::Parser;
 use futures::{Stream, StreamExt, TryStreamExt};
-use sqlx::PgPool;
+use sqlx::{postgres::PgPoolOptions, PgPool};
 use tap::{Tap, TapFallible, TapOptional};
 use tendermint::abci;
 use tracing::{debug, info};
@@ -16,9 +15,9 @@ pub struct Indexer {
 }
 
 impl Indexer {
-    pub fn new() -> Self {
+    pub fn new(opts: Options) -> Self {
         Self {
-            opts: Options::parse(),
+            opts,
             indexes: Vec::new(),
         }
     }
@@ -60,7 +59,22 @@ impl Indexer {
             indexes,
         } = self;
 
-        let src_db = PgPool::connect(&src_database_url).await?;
+        // Create a source db, with, for sanity, some read only settings.
+        // These will be overrideable by a consumer who knows what they're doing,
+        // but prevents basic mistakes.
+        // c.f. https://github.com/launchbadge/sqlx/issues/481#issuecomment-727011811
+        let src_db = PgPoolOptions::new()
+            .after_connect(|conn, _| {
+                Box::pin(async move {
+                    sqlx::query("SET SESSION CHARACTERISTICS AS TRANSACTION READ ONLY;")
+                        .execute(conn)
+                        .await?;
+                    Ok(())
+                })
+            })
+            .connect(&src_database_url)
+            .await?;
+
         let dst_db = PgPool::connect(&dst_database_url).await?;
 
         // Check if the destination db is initialized
@@ -169,7 +183,7 @@ impl Indexer {
             for index in indexes {
                 if index.is_relevant(&event.as_ref().kind) {
                     tracing::debug!(?event, ?index, "relevant to index");
-                    index.index_event(&mut dbtx, &event).await?;
+                    index.index_event(&mut dbtx, &event, &src_db).await?;
                 }
             }
             // Mark that we got to at least this event

crates/util/cometindex/src/lib.rs

@@ -5,7 +5,7 @@ pub mod indexer;
 pub mod opt;
 
 pub use contextualized::ContextualizedEvent;
-pub use index::{AppView, PgTransaction};
+pub use index::{AppView, PgPool, PgTransaction};
 pub use indexer::Indexer;
 
 pub use async_trait::async_trait;

crates/util/cometindex/src/opt.rs

@@ -4,7 +4,7 @@ use anyhow::{Error, Result};
 use clap::Parser;
 
 /// This struct represents the command-line options
-#[derive(Debug, Parser)]
+#[derive(Clone, Debug, Parser)]
 #[clap(
     name = "cometindex",
     about = "processes raw events emitted by cometbft applications",