pindexer: repair SQL syntax for create type/create domain (#4760)

Fixes some SQL statements in the prior commit which were not valid
postgres.

- [ ] 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:

  > Only indexer changes.

---------

Co-authored-by: Conor Schaefer <[email protected]>
Co-authored-by: Lucas Meier <[email protected]>

crates/bin/pindexer/src/dex/dex.sql

@@ -9,13 +9,17 @@
 -- that given an `penumbra_asset::asset::Id`, we always know exactly how to filter
 -- tables, rather than needing to do a join with another table.
 
-CREATE DOMAIN IF NOT EXISTS Amount AS NUMERIC(39, 0) NOT NULL;
 
+DROP TYPE IF EXISTS Value CASCADE;
+DROP DOMAIN IF EXISTS Amount;
+
+CREATE DOMAIN Amount AS NUMERIC(39, 0) NOT NULL;
 CREATE TYPE Value AS (
   amount Amount,
-  asset BYTEA NOT NULL
+  asset BYTEA
 );
 
+
 -- Keeps track of changes to the dex's value circuit breaker.
 CREATE TABLE IF NOT EXISTS dex_value_circuit_breaker_change (
   -- The asset being moved into or out of the dex.
@@ -29,21 +33,21 @@ CREATE TABLE IF NOT EXISTS dex_value_circuit_breaker_change (
 -- One step of an execution trace.
 CREATE TABLE IF NOT EXISTS trace_step (
   id SERIAL PRIMARY KEY,
-  value Value,
+  value Value
 );
 
 -- A single trace, showing what a small amount of an input asset was exchanged for.
 CREATE TABLE IF NOT EXISTS trace (
   id SERIAL PRIMARY KEY,
   step_start INTEGER REFERENCES trace_step(id),
-  step_end INTEGER REFERENCES trace_step(id),
+  step_end INTEGER REFERENCES trace_step(id)
 );
 
 --- Represents instances where arb executions happened.
 CREATE TABLE IF NOT EXISTS arb (
   height BIGINT PRIMARY KEY,
   input Value,
   output Value,
-  trace_start INTEGER REFERENCES arb_traces(id),
-  trace_end INTEGER REFERENCES arb_traces(id),
+  trace_start INTEGER REFERENCES trace(id),
+  trace_end INTEGER REFERENCES trace(id)
 );

crates/bin/pindexer/src/dex/mod.rs

@@ -207,9 +207,9 @@ impl AppView for Component {
         dbtx: &mut PgTransaction,
         _app_state: &serde_json::Value,
     ) -> anyhow::Result<()> {
-        sqlx::query(include_str!("dex.sql"))
-            .execute(dbtx.as_mut())
-            .await?;
+        for statement in include_str!("dex.sql").split(";") {
+            sqlx::query(statement).execute(dbtx.as_mut()).await?;
+        }
         Ok(())
     }
 

flake.nix

@@ -129,6 +129,7 @@
               protobuf
               rocksdb
               rsync
+              sqlfluff
             ];
             shellHook = ''
               export LIBCLANG_PATH=${LIBCLANG_PATH}