@paimaexample/runtime@0.3.5

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586
import { type ChainBlock, genSyncProtocols } from "jsr:@paimaexample/sync@^0.3.5";
import {
  aquireDBMutex,
  createDynamicTables,
  getConnection,
  releaseDBMutex,
} from "jsr:@paimaexample/db@^0.3.5";
import { startMerge, startSync } from "jsr:@paimaexample/sync@^0.3.5";
import { ComponentNames, log, SeverityNumber } from "jsr:@paimaexample/log@^0.3.5";
import { createChannel, each, type Operation, spawn, until } from "npm:effection@^3.5.0";
import { initTelemetry } from "./telemetry.ts";
import { processFinalizedBlock } from "./process-blocks.ts";
import { startHttpServer } from "./api/http-server.ts";
import type { StartConfig } from "./types.ts";
import type { Client } from "npm:pg@^8.14.0";
import type { PaimaBlockHash } from "jsr:@paimaexample/utils@^0.3.5";

export function* init() {
  // initialize OpenTelemetry
  yield* initTelemetry();
}

/**
 * Main entry point to start the Paima Engine Node.
 *
 * This will launch the networks/primitives syncronization sub-processes,
 * the HTTP server, and the merge and paima-block generation process.
 *
 * @param config - Paima Engine Node configuration object.
 */
export function* start(config: StartConfig): Operation<void> {
  const { syncInfo } = config;
  const dbConn = getConnection();
  const syncProtocols = yield* genSyncProtocols(dbConn, syncInfo);

  // TODO We only need to do this once, at the beginning.
  //      We have to distinguish between the start or restart of the node.
  //      Futher updates need to be managed by the user.
  yield* createDynamicTables(dbConn, syncProtocols);

  log.remote(
    ComponentNames.PAIMA_RUNTIME,
    [],
    SeverityNumber.INFO,
    (log) => log("start sync"),
  );
  for (const syncProtocol of syncProtocols) {
    yield* startSync(syncProtocol);
  }

  yield* spawn(function* () {
    yield* startHttpServer(dbConn, syncProtocols, config.apiRouter);
  });

  const finalizedBlockStream = createChannel<ChainBlock>();

  yield* spawn(() => startMerge(syncProtocols, finalizedBlockStream));

  let blockHash: PaimaBlockHash | null = null;
  for (const value of yield* each(finalizedBlockStream)) {
    // We request a dbClient for a non-shared dbConn object.
    // For PGLite, this is not enough, as the can only be one connection at a time.
    // So we request a DBMutex as well.
    const dbClient: Client = yield* until(dbConn.connect());
    try {
      yield* aquireDBMutex("processing-blocks");
      blockHash = yield* processFinalizedBlock(
        value,
        config,
        dbClient,
        blockHash,
      );
    } finally {
      releaseDBMutex();
      dbClient.release();
    }

    log.remote(
      ComponentNames.PAIMA_SYNC,
      "block-merge",
      SeverityNumber.INFO,
      (log) => log(`finalized block ${value.blockNumber} @ ${blockHash}`),
    );
    yield* each.next();
  }
}