add task that tracks stake delegation

indexer/entity/src/lib.rs

@@ -15,4 +15,5 @@ pub mod dex_swap;
 pub mod native_asset;
 pub mod plutus_data;
 pub mod plutus_data_hash;
+pub mod stake_delegation;
 pub mod transaction_metadata;

indexer/entity/src/stake_delegation.rs

@@ -0,0 +1,37 @@
+use sea_orm::entity::prelude::*;
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel, Deserialize, Serialize)]
+#[sea_orm(table_name = "StakeDelegationCredentialRelation")]
+pub struct Model {
+    #[sea_orm(primary_key, column_type = "BigInteger")]
+    pub id: i64,
+    pub stake_credential: i64,
+    // pool registrations are not tracked in StakeCredentials,
+    pub pool_credential: Option<Vec<u8>>,
+    pub tx_id: i64,
+}
+
+#[derive(Copy, Clone, Debug, DeriveRelation, EnumIter)]
+pub enum Relation {
+    #[sea_orm(
+        belongs_to = "super::stake_credential::Entity",
+        from = "Column::StakeCredential",
+        to = "super::stake_credential::Column::Id"
+    )]
+    StakeCredential,
+    #[sea_orm(
+        belongs_to = "super::transaction::Entity",
+        from = "Column::TxId",
+        to = "super::transaction::Column::Id"
+    )]
+    Transaction,
+}
+
+impl Related<super::stake_credential::Entity> for Entity {
+    fn to() -> RelationDef {
+        Relation::StakeCredential.def()
+    }
+}
+
+impl ActiveModelBehavior for ActiveModel {}

indexer/execution_plans/default.toml

@@ -63,3 +63,5 @@ readonly=false
 readonly=false
 
 [MultieraCip25EntryTask]
+
+[MultieraAddressDelegationTask]

indexer/migration/src/lib.rs

@@ -17,6 +17,7 @@ mod m20220528_000012_create_plutus_data_table;
 mod m20220808_000013_create_transaction_reference_input_table;
 mod m20221031_000014_create_dex_table;
 mod m20230223_000015_modify_block_table;
+mod m20230927_231206_create_stake_delegation_table;
 
 pub struct Migrator;
 
@@ -41,6 +42,7 @@ impl MigratorTrait for Migrator {
             Box::new(m20220808_000013_create_transaction_reference_input_table::Migration),
             Box::new(m20221031_000014_create_dex_table::Migration),
             Box::new(m20230223_000015_modify_block_table::Migration),
+            Box::new(m20230927_231206_create_stake_delegation_table::Migration),
         ]
     }
 }

indexer/migration/src/m20230927_231206_create_stake_delegation_table.rs

@@ -0,0 +1,74 @@
+use entity::prelude::{StakeCredential, StakeCredentialColumn, Transaction, TransactionColumn};
+use entity::stake_delegation::*;
+use sea_schema::migration::prelude::*;
+
+pub struct Migration;
+
+impl MigrationName for Migration {
+    fn name(&self) -> &str {
+        "m20230927_231206_create_stake_delegation_table"
+    }
+}
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .create_table(
+                Table::create()
+                    .table(Entity)
+                    .if_not_exists()
+                    .col(
+                        ColumnDef::new(Column::Id)
+                            .big_integer()
+                            .not_null()
+                            .auto_increment(),
+                    )
+                    .col(
+                        ColumnDef::new(Column::StakeCredential)
+                            .big_integer()
+                            .not_null(),
+                    )
+                    .col(ColumnDef::new(Column::TxId).big_integer().not_null())
+                    .col(ColumnDef::new(Column::PoolCredential).binary())
+                    .foreign_key(
+                        ForeignKey::create()
+                            .name("fk-stake_delegation-credential_id")
+                            .from(Entity, Column::StakeCredential)
+                            .to(StakeCredential, StakeCredentialColumn::Id)
+                            .on_delete(ForeignKeyAction::Cascade),
+                    )
+                    .foreign_key(
+                        ForeignKey::create()
+                            .name("fk-stake_delegation-tx_id")
+                            .from(Entity, Column::TxId)
+                            .to(Transaction, TransactionColumn::Id)
+                            .on_delete(ForeignKeyAction::Cascade),
+                    )
+                    .primary_key(
+                        Index::create()
+                            .table(Entity)
+                            .name("stake_delegation_credential-pk")
+                            .col(Column::Id),
+                    )
+                    .to_owned(),
+            )
+            .await
+
+        // manager
+        //     .create_index(
+        //         Index::create()
+        //             .table(Entity)
+        //             .name("index-address_credential-credential")
+        //             .col(Column::CredentialId)
+        //             .to_owned(),
+        //     )
+        //     .await
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .drop_table(Table::drop().table(Entity).to_owned())
+            .await
+    }
+}

indexer/tasks/src/multiera/mod.rs

@@ -1,6 +1,7 @@
 pub mod dex;
 pub mod multiera_address;
 pub mod multiera_address_credential_relations;
+pub mod multiera_address_delegation;
 pub mod multiera_asset_mint;
 pub mod multiera_block;
 pub mod multiera_cip25entry;

indexer/tasks/src/multiera/multiera_address_delegation.rs

@@ -0,0 +1,113 @@
+use crate::{
+    multiera::multiera_stake_credentials::MultieraStakeCredentialTask,
+    types::{AddressCredentialRelationValue, TxCredentialRelationValue},
+};
+use cardano_multiplatform_lib::{
+    address::{BaseAddress, EnterpriseAddress, PointerAddress, RewardAddress},
+    byron::ByronAddress,
+};
+use entity::{
+    prelude::*,
+    sea_orm::{prelude::*, DatabaseTransaction},
+};
+use pallas::ledger::{
+    primitives::{alonzo::Certificate, Fragment},
+    traverse::{MultiEraBlock, MultiEraCert, MultiEraOutput, MultiEraTx},
+};
+use sea_orm::{Order, QueryOrder, Set};
+use std::collections::{BTreeMap, BTreeSet};
+use std::ops::Deref;
+
+use super::{
+    multiera_address_credential_relations::QueuedAddressCredentialRelation,
+    multiera_txs::MultieraTransactionTask, relation_map::RelationMap,
+};
+use crate::config::EmptyConfig::EmptyConfig;
+use crate::dsl::database_task::BlockGlobalInfo;
+use crate::dsl::task_macro::*;
+
+carp_task! {
+  name MultieraAddressDelegationTask;
+  configuration EmptyConfig;
+  doc "Tracks stake delegation actions to pools.";
+  era multiera;
+  dependencies [MultieraStakeCredentialTask];
+  read [multiera_txs, multiera_stake_credential];
+  write [];
+  should_add_task |block, _properties| {
+    // recall: txs may have no outputs if they just burn all inputs as fee
+    // TODO: this runs slightly more than it should
+    !block.1.is_empty()
+  };
+  execute |previous_data, task| handle(
+      task.db_tx,
+      task.block,
+      &previous_data.multiera_txs,
+      &previous_data.multiera_stake_credential,
+  );
+  merge_result |_previous_data, _result| {};
+}
+
+async fn handle(
+    db_tx: &DatabaseTransaction,
+    block: BlockInfo<'_, MultiEraBlock<'_>, BlockGlobalInfo>,
+    multiera_txs: &[TransactionModel],
+    multiera_stake_credential: &BTreeMap<Vec<u8>, StakeCredentialModel>,
+) -> Result<(), DbErr> {
+    for (tx_body, cardano_transaction) in block.1.txs().iter().zip(multiera_txs) {
+        for cert in tx_body.certs() {
+            {
+                let tx_id = cardano_transaction.id;
+                let cert = &cert;
+                match cert.as_alonzo().unwrap() {
+                    Certificate::StakeDelegation(credential, pool) => {
+                        let credential = credential.encode_fragment().unwrap();
+
+                        let stake_credential_id = multiera_stake_credential
+                            .get(&credential.to_vec())
+                            .unwrap()
+                            .id;
+
+                        entity::stake_delegation::ActiveModel {
+                            stake_credential: Set(stake_credential_id),
+                            pool_credential: Set(Some(pool.to_vec())),
+                            tx_id: Set(tx_id),
+                            ..Default::default()
+                        }
+                        .save(db_tx)
+                        .await?;
+                    }
+                    Certificate::StakeRegistration(credential) => {}
+                    Certificate::StakeDeregistration(credential) => {
+                        let credential = credential.encode_fragment().unwrap();
+
+                        let stake_credential_id = multiera_stake_credential
+                            .get(&credential.to_vec())
+                            .unwrap()
+                            .id;
+
+                        entity::stake_delegation::ActiveModel {
+                            stake_credential: Set(stake_credential_id),
+                            pool_credential: Set(None),
+                            tx_id: Set(tx_id),
+                            ..Default::default()
+                        }
+                        .save(db_tx)
+                        .await?;
+                    }
+                    Certificate::PoolRegistration {
+                        operator,
+                        pool_owners,
+                        reward_account,
+                        ..
+                    } => {}
+                    Certificate::PoolRetirement(key_hash, _) => {}
+                    Certificate::GenesisKeyDelegation(_, _, _) => {}
+                    Certificate::MoveInstantaneousRewardsCert(mir) => {}
+                };
+            };
+        }
+    }
+
+    Ok(())
+}

webserver/server/app/controllers/DelegationForAddressController.ts

@@ -0,0 +1,59 @@
+import { Body, Controller, TsoaResponse, Res, Post, Route, SuccessResponse } from 'tsoa';
+import { StatusCodes } from 'http-status-codes';
+import tx from 'pg-tx';
+import pool from '../services/PgPoolSingleton';
+import type { ErrorShape } from '../../../shared/errors';
+import { genErrorMessage } from '../../../shared/errors';
+import { Errors } from '../../../shared/errors';
+import type { EndpointTypes } from '../../../shared/routes';
+import { Routes } from '../../../shared/routes';
+import { getAddressTypes } from '../models/utils';
+import { delegationForAddress } from '../services/Delegation';
+import { DelegationForAddressResponse } from '../../../shared/models/DelegationForAddress';
+
+const route = Routes.delegationForAddress;
+
+@Route('delegation/address')
+export class DelegationForAddressController extends Controller {
+    @SuccessResponse(`${StatusCodes.OK}`)
+    @Post()
+    public async delegationForAddress(
+        @Body()
+        requestBody: EndpointTypes[typeof route]['input'],
+        @Res()
+        errorResponse: TsoaResponse<
+            StatusCodes.BAD_REQUEST | StatusCodes.CONFLICT | StatusCodes.UNPROCESSABLE_ENTITY,
+            ErrorShape
+        >
+    ): Promise<EndpointTypes[typeof route]['response']> {
+        const addressTypes = getAddressTypes([requestBody.address]);
+
+        if (addressTypes.invalid.length > 0) {
+            // eslint-disable-next-line @typescript-eslint/no-unsafe-return
+            return errorResponse(
+                StatusCodes.UNPROCESSABLE_ENTITY,
+                genErrorMessage(Errors.IncorrectAddressFormat, {
+                    addresses: addressTypes.invalid,
+                })
+            );
+        }
+
+        const response = await tx<
+            DelegationForAddressResponse
+        >(pool, async dbTx => {
+            const data = await delegationForAddress({
+                address: addressTypes.credentialHex.map(addr => Buffer.from(addr, 'hex'))[0],
+                until: requestBody.until,
+                dbTx
+            });
+
+            return {
+                pool: data ? data.pool : null,
+                txId: data ? data.tx_id : null,
+            }
+        });
+
+        return response;
+    }
+}
+

webserver/server/app/models/delegation/delegationForAddress.queries.ts

@@ -0,0 +1,41 @@
+/** Types generated for queries found in "app/models/delegation/delegationForAddress.sql" */
+import { PreparedQuery } from '@pgtyped/query';
+
+/** 'SqlStakeDelegation' parameters type */
+export interface ISqlStakeDelegationParams {
+  credential: Buffer;
+  slot: number;
+}
+
+/** 'SqlStakeDelegation' return type */
+export interface ISqlStakeDelegationResult {
+  pool: string | null;
+  tx_id: string | null;
+}
+
+/** 'SqlStakeDelegation' query type */
+export interface ISqlStakeDelegationQuery {
+  params: ISqlStakeDelegationParams;
+  result: ISqlStakeDelegationResult;
+}
+
+const sqlStakeDelegationIR: any = {"usedParamSet":{"credential":true,"slot":true},"params":[{"name":"credential","required":true,"transform":{"type":"scalar"},"locs":[{"a":371,"b":382}]},{"name":"slot","required":true,"transform":{"type":"scalar"},"locs":[{"a":405,"b":410}]}],"statement":"SELECT encode(pool_credential, 'hex') as pool, encode(\"Transaction\".hash, 'hex') as tx_id\nFROM \"StakeDelegationCredentialRelation\"\nJOIN \"StakeCredential\" ON stake_credential = \"StakeCredential\".id\nJOIN \"Transaction\" ON \"Transaction\".id = \"StakeDelegationCredentialRelation\".tx_id\nJOIN \"Block\" ON \"Transaction\".block_id = \"Block\".id\nWHERE \n\t\"StakeCredential\".credential = :credential! AND\n\t\"Block\".slot <= :slot!\nORDER BY (\"Block\".height, \"Transaction\".tx_index) DESC\nLIMIT 1"};
+
+/**
+ * Query generated from SQL:
+ * ```
+ * SELECT encode(pool_credential, 'hex') as pool, encode("Transaction".hash, 'hex') as tx_id
+ * FROM "StakeDelegationCredentialRelation"
+ * JOIN "StakeCredential" ON stake_credential = "StakeCredential".id
+ * JOIN "Transaction" ON "Transaction".id = "StakeDelegationCredentialRelation".tx_id
+ * JOIN "Block" ON "Transaction".block_id = "Block".id
+ * WHERE 
+ * 	"StakeCredential".credential = :credential! AND
+ * 	"Block".slot <= :slot!
+ * ORDER BY ("Block".height, "Transaction".tx_index) DESC
+ * LIMIT 1
+ * ```
+ */
+export const sqlStakeDelegation = new PreparedQuery<ISqlStakeDelegationParams,ISqlStakeDelegationResult>(sqlStakeDelegationIR);
+
+

webserver/server/app/models/delegation/delegationForAddress.sql

@@ -0,0 +1,11 @@
+/* @name sqlStakeDelegation */
+SELECT encode(pool_credential, 'hex') as pool, encode("Transaction".hash, 'hex') as tx_id
+FROM "StakeDelegationCredentialRelation"
+JOIN "StakeCredential" ON stake_credential = "StakeCredential".id
+JOIN "Transaction" ON "Transaction".id = "StakeDelegationCredentialRelation".tx_id
+JOIN "Block" ON "Transaction".block_id = "Block".id
+WHERE 
+	"StakeCredential".credential = :credential! AND
+	"Block".slot <= :slot!
+ORDER BY ("Block".height, "Transaction".tx_index) DESC
+LIMIT 1;

webserver/server/app/services/Delegation.ts

@@ -0,0 +1,12 @@
+
+import type { PoolClient } from 'pg';
+import { ISqlStakeDelegationResult, sqlStakeDelegation } from '../models/delegation/delegationForAddress.queries';
+
+
+export async function delegationForAddress(request: {
+    address: Buffer,
+    until: { absoluteSlot: number },
+    dbTx: PoolClient,
+}): Promise<ISqlStakeDelegationResult> {
+    return (await sqlStakeDelegation.run({ credential: request.address, slot: request.until.absoluteSlot }, request.dbTx))[0];
+}