Merge pull request #680 from input-output-hk/feat/pool-metadata-task-…

…queue

feat: pool metadata task queue / single-tenant projection support

demo/README.md

@@ -0,0 +1,27 @@
+## [projection-typeorm.js](./projection-typeorm.js)
+
+An example of [projection](../packages/projection/) into PostgreSQL database ([projection-typeorm](../packages/projection-typeorm/)).
+
+### Environment
+
+```sh
+cd /path/to/cardano-js-sdk/ # monorepo root
+yarn && yarn build
+yarn preprod:up cardano-node-ogmios postgres # or preview:up/mainnet:up
+```
+
+### Configuration
+
+Projection can be customized by adding/removing operators and TypeORM entities, e.g. you may add those operators `Mappers.withStakeKeys(), ..., storeStakeKeys()` alongside with adding a `StakeKeyEntity`.
+
+### Running the script
+
+```sh
+node demo/projection-typeorm.js
+```
+
+### Recreating the schema
+
+```sh
+node demo/projection-typeorm.js --drop
+```

demo/projection-typeorm.js

@@ -0,0 +1,131 @@
+// Runtime dependency: `yarn preprod:up cardano-node-ogmios postgres` (can be any network)
+/* eslint-disable import/no-extraneous-dependencies */
+const { Bootstrap, Mappers, requestNext, logProjectionProgress } = require('@cardano-sdk/projection');
+const {
+  createDataSource,
+  withTypeormTransaction,
+  typeormTransactionCommit,
+  TypeormStabilityWindowBuffer,
+  BlockDataEntity,
+  BlockEntity,
+  StakePoolEntity,
+  PoolRegistrationEntity,
+  PoolRetirementEntity,
+  OutputEntity,
+  AssetEntity,
+  TokensEntity,
+  storeBlock,
+  storeAssets,
+  storeUtxo,
+  storeStakePools,
+  storeStakePoolMetadataJob,
+  isRecoverableTypeormError
+} = require('@cardano-sdk/projection-typeorm');
+const { OgmiosObservableCardanoNode } = require('@cardano-sdk/ogmios');
+const { defer, from, of } = require('rxjs');
+const { createDatabase } = require('typeorm-extension');
+const { shareRetryBackoff } = require('@cardano-sdk/util-rxjs');
+
+const logger = {
+  ...console,
+  debug: () => void 0
+};
+
+const entities = [
+  BlockDataEntity,
+  BlockEntity,
+  StakePoolEntity,
+  PoolRegistrationEntity,
+  PoolRetirementEntity,
+  AssetEntity,
+  TokensEntity,
+  OutputEntity
+];
+const extensions = {
+  pgBoss: true
+};
+
+const cardanoNode = new OgmiosObservableCardanoNode(
+  {
+    connectionConfig$: of({
+      port: 1339
+    })
+  },
+  { logger }
+);
+const buffer = new TypeormStabilityWindowBuffer({ logger });
+
+// #region TypeORM setup
+
+const connectionConfig = {
+  database: 'projection',
+  host: 'localhost',
+  password: 'doNoUseThisSecret!',
+  username: 'postgres'
+};
+
+const dataSource$ = defer(() =>
+  from(
+    (async () => {
+      const dataSource = createDataSource({
+        connectionConfig,
+        devOptions: process.argv.includes('--drop')
+          ? {
+              dropSchema: true,
+              synchronize: true
+            }
+          : undefined,
+        entities,
+        extensions,
+        logger
+      });
+      await createDatabase({
+        options: {
+          type: 'postgres',
+          ...connectionConfig,
+          installExtensions: true
+        }
+      });
+      await dataSource.initialize();
+      await buffer.initialize(dataSource.createQueryRunner());
+      return dataSource;
+    })()
+  )
+);
+
+// #endregion
+
+Bootstrap.fromCardanoNode({
+  buffer,
+  cardanoNode,
+  logger
+})
+  .pipe(
+    Mappers.withCertificates(),
+    Mappers.withStakePools(),
+    Mappers.withMint(),
+    Mappers.withUtxo(),
+    // Single-tenant example
+    // Mappers.filterProducedUtxoByAddresses({
+    //   addresses: [
+    //     'addr_test1qpgn04xka0857kh6859za75tfvlrlu2lft0yc9z87598yjezw8yvpkv977yj5va20xmd9vw5fczfl3uu4expskz8adfqpydths'
+    //   ]
+    // }),
+    shareRetryBackoff(
+      (evt$) =>
+        evt$.pipe(
+          withTypeormTransaction({ dataSource$, logger }, extensions),
+          storeBlock(),
+          buffer.storeBlockData(),
+          storeAssets(),
+          storeUtxo(),
+          storeStakePools(),
+          storeStakePoolMetadataJob(),
+          typeormTransactionCommit()
+        ),
+      { shouldRetry: isRecoverableTypeormError }
+    ),
+    requestNext(),
+    logProjectionProgress(logger)
+  )
+  .subscribe();

packages/e2e/package.json

@@ -162,6 +162,7 @@
     "readable-stream": "^3.6.0",
     "shx": "^0.3.3",
     "ts-jest": "^28.0.7",
+    "typeorm": "^0.3.12",
     "typeorm-extension": "^2.5.2",
     "typescript": "^4.7.4",
     "util": "^0.12.4",