refactor!: simplify projection Sink to be an operator

Instead of {sink,before,after}, a Sink is now just an rxjs operator.
It gives more control to the sink implementation - it can now decide
the order of granular sinks. It is needed for stakePoolMetadata,
which requires stakePool to be run before it.

Also extract the bootstrapping part from projectIntoSink, which is
now a separate export `Bootstrap.fromCardanoNode`. It simplifies
projectIntoSink function by decoupling it form ObservableCardanoNode.
This refactor wasn't strictly required right now, but projectIntoSink
was previously too large and difficult to work with.

Lastly, remove StabilityWindowBuffer.handleEvents and only keep
tip$ and tail$, leaving the mechanism of updating the buffer up to
concrete implementations. It is no longer quired since handleEvents
is no longer called by projectIntoSink.

Author: mkazlauskas

packages/e2e/test/projection/offline-fork.test.ts

@@ -1,5 +1,14 @@
 /* eslint-disable promise/always-return */
 import * as Postgres from '@cardano-sdk/projection-typeorm';
+import {
+  Bootstrap,
+  InMemory,
+  Projections,
+  Sink,
+  StabilityWindowBuffer,
+  WithBlock,
+  projectIntoSink
+} from '@cardano-sdk/projection';
 import {
   Cardano,
   ChainSyncEvent,
@@ -10,7 +19,6 @@ import {
 } from '@cardano-sdk/core';
 import { ChainSyncDataSet, chainSyncData, logger } from '@cardano-sdk/util-dev';
 import { ConnectionConfig } from '@cardano-ogmios/client';
-import { InMemory, Projections, SinksFactory, WithBlock, projectIntoSink } from '@cardano-sdk/projection';
 import { Observable, filter, firstValueFrom, lastValueFrom, of, share, take } from 'rxjs';
 import { OgmiosObservableCardanoNode } from '@cardano-sdk/ogmios';
 import { createDatabase } from 'typeorm-extension';
@@ -108,35 +116,36 @@ describe('resuming projection when intersection is not local tip', () => {
     ogmiosCardanoNode = new OgmiosObservableCardanoNode({ connectionConfig$: of(ogmiosConnectionConfig) }, { logger });
   });
 
-  const project = (cardanoNode: ObservableCardanoNode, sinksFactory: SinksFactory<typeof projections>) =>
+  const project = (cardanoNode: ObservableCardanoNode, buffer: StabilityWindowBuffer, sink: Sink<typeof projections>) =>
     projectIntoSink({
-      cardanoNode,
       logger,
       projections,
-      sinksFactory
+      sink,
+      source$: Bootstrap.fromCardanoNode({ buffer, cardanoNode, logger })
     });
 
   const testRollbackAndContinue = (
-    sinksFactory: SinksFactory<typeof projections>,
+    buffer: StabilityWindowBuffer,
+    sink: Sink<typeof projections>,
     getNumberOfLocalStakeKeys: () => Promise<number>
   ) => {
     it('rolls back local data to intersection and resumes projection from there', async () => {
       // Project some events until we find at least 1 stake key registration
       const firstEventWithKeyRegistrations = await firstValueFrom(
-        project(ogmiosCardanoNode, sinksFactory).pipe(filter((evt) => evt.stakeKeys.insert.length > 0))
+        project(ogmiosCardanoNode, buffer, sink).pipe(filter((evt) => evt.stakeKeys.insert.length > 0))
       );
       const lastEventFromOriginalSync = firstEventWithKeyRegistrations;
       const numStakeKeysBeforeFork = await getNumberOfLocalStakeKeys();
       expect(numStakeKeysBeforeFork).toBe(firstEventWithKeyRegistrations.stakeKeys.insert.length); // sanity check
 
       // Simulate a fork by adding some blocks that are not on the ogmios chain
       const stubForkCardanoNode = createForkProjectionSource(ogmiosCardanoNode, lastEventFromOriginalSync);
-      await lastValueFrom(project(stubForkCardanoNode, sinksFactory).pipe(take(4)));
+      await lastValueFrom(project(stubForkCardanoNode, buffer, sink).pipe(take(4)));
       const numStakeKeysAfterFork = await getNumberOfLocalStakeKeys();
       expect(numStakeKeysAfterFork).toBeGreaterThan(numStakeKeysBeforeFork);
 
       // Continue projection from ogmios
-      const continue$ = project(ogmiosCardanoNode, sinksFactory).pipe(share());
+      const continue$ = project(ogmiosCardanoNode, buffer, sink).pipe(share());
       const rollForward$ = continue$.pipe(filter((evt) => evt.eventType === ChainSyncEventType.RollForward));
       const rolledBackKeyRegistrations$ = continue$.pipe(
         filter(
@@ -178,14 +187,13 @@ describe('resuming projection when intersection is not local tip', () => {
 
   describe('InMemory', () => {
     const store = InMemory.createStore();
-    const sinks = InMemory.createSinks(store);
-    testRollbackAndContinue(
-      () => sinks,
-      async () => store.stakeKeys.size
-    );
+    const buffer = new InMemory.InMemoryStabilityWindowBuffer();
+    const sink = InMemory.createSink(store, buffer);
+    testRollbackAndContinue(buffer, sink, async () => store.stakeKeys.size);
   });
 
   describe('typeorm', () => {
+    const buffer = new Postgres.TypeormStabilityWindowBuffer({ logger });
     const dataSource = Postgres.createDataSource({
       connectionConfig: pgConnectionConfig,
       devOptions: {
@@ -198,7 +206,8 @@ describe('resuming projection when intersection is not local tip', () => {
       },
       projections
     });
-    const sinksFactory = Postgres.createSinksFactory({
+    const sink = Postgres.createSink({
+      buffer,
       dataSource$: of(dataSource),
       logger
     });
@@ -219,6 +228,6 @@ describe('resuming projection when intersection is not local tip', () => {
     });
     afterAll(() => dataSource.destroy());
 
-    testRollbackAndContinue(sinksFactory, getNumberOfLocalStakeKeys);
+    testRollbackAndContinue(buffer, sink, getNumberOfLocalStakeKeys);
   });
 });

packages/projection-typeorm/src/TypeormStabilityWindowBuffer.ts

@@ -23,7 +23,7 @@ const blockDataSelect: FindOptionsSelect<BlockDataEntity> = {
   data: true as any
 };
 
-export interface TypeormStabilityWindowBufferProps {
+export interface TypeormStabilityWindowBufferProps extends WithLogger {
   /**
    * 100 by default
    */
@@ -34,32 +34,32 @@ export interface TypeormStabilityWindowBufferProps {
   allowNonSequentialBlockHeights?: boolean;
 }
 
-export class TypeormStabilityWindowBuffer
-  implements StabilityWindowBuffer<Operators.WithNetworkInfo & WithTypeormContext>
-{
+export class TypeormStabilityWindowBuffer implements StabilityWindowBuffer {
   #tail: Cardano.Block | 'origin';
+  #tip$: ReplaySubject<Cardano.Block | 'origin'>;
+  #tail$: ReplaySubject<Cardano.Block | 'origin'>;
+  tip$: Observable<Cardano.Block | 'origin'>;
+  tail$: Observable<Cardano.Block | 'origin'>;
   readonly #logger: Logger;
   readonly #compactEvery: number;
   readonly #allowNonSequentialBlockHeights?: boolean;
-  readonly #tip$ = new ReplaySubject<Cardano.Block | 'origin'>(1);
-  readonly #tail$ = new ReplaySubject<Cardano.Block | 'origin'>(1);
-  readonly tip$: Observable<Cardano.Block | 'origin'>;
-  readonly tail$: Observable<Cardano.Block | 'origin'>;
-  readonly handleEvents: UnifiedProjectorOperator<
-    Operators.WithNetworkInfo & WithTypeormContext,
-    Operators.WithNetworkInfo & WithTypeormContext
-  >;
 
-  constructor(
-    { allowNonSequentialBlockHeights, compactBufferEveryNBlocks = 100 }: TypeormStabilityWindowBufferProps,
-    dependencies: WithLogger
-  ) {
-    this.tip$ = this.#tip$.asObservable();
-    this.tail$ = this.#tail$.asObservable();
+  constructor({
+    allowNonSequentialBlockHeights,
+    compactBufferEveryNBlocks = 100,
+    logger
+  }: TypeormStabilityWindowBufferProps) {
+    this.start();
     this.#compactEvery = compactBufferEveryNBlocks;
     this.#allowNonSequentialBlockHeights = allowNonSequentialBlockHeights;
-    this.#logger = contextLogger(dependencies.logger, 'PgStabilityWindowBuffer');
-    this.handleEvents = (evt$) =>
+    this.#logger = contextLogger(logger, 'TypeormStabilityWindowBuffer');
+  }
+
+  handleEvents(): UnifiedProjectorOperator<
+    Operators.WithNetworkInfo & WithTypeormContext,
+    Operators.WithNetworkInfo & WithTypeormContext
+  > {
+    return (evt$) =>
       evt$.pipe(
         concatMap((evt) =>
           from(
@@ -86,6 +86,18 @@ export class TypeormStabilityWindowBuffer
     this.#setTail(tail[0]?.data || 'origin');
   }
 
+  /**
+   * (Re)create tip$ and tail$ observables.
+   * Automatically called in constructor.
+   * Can be used to reuse the same buffer object after shutdown().
+   */
+  start() {
+    this.#tip$ = new ReplaySubject(1);
+    this.#tail$ = new ReplaySubject(1);
+    this.tip$ = this.#tip$.asObservable();
+    this.tail$ = this.#tail$.asObservable();
+  }
+
   shutdown(): void {
     this.#tip$.complete();
     this.#tail$.complete();

packages/projection-typeorm/src/createDataSource.ts

@@ -1,10 +1,9 @@
 import 'reflect-metadata';
-import * as supportedSinks from './sinks';
 import { BlockDataEntity } from './entity';
 import { BlockEntity } from './entity/Block.entity';
 import { DataSource, DataSourceOptions, DefaultNamingStrategy, NamingStrategyInterface } from 'typeorm';
 import { Logger } from 'ts-log';
-import { WithTypeormSinkMetadata } from './types';
+import { supportedSinks } from './util';
 import { typeormLogger } from './logger';
 import snakeCase from 'lodash/snakeCase';
 import uniq from 'lodash/uniq';
@@ -93,7 +92,7 @@ export const createDataSource = <P extends object>({
   projections,
   logger
 }: CreateDataSourceProps<P>) => {
-  const requestedProjectionEntities = Object.entries<WithTypeormSinkMetadata>(supportedSinks)
+  const requestedProjectionEntities = Object.entries(supportedSinks)
     .filter(([projectionName]) => projectionName in projections)
     .flatMap(([_, sink]) => sink.entities);
   const entities: Function[] = uniq([BlockEntity, BlockDataEntity, ...requestedProjectionEntities]);