feat(data): add contiguous sequntial data source PE-2810

Tries contiguous data source one at a time until it finds one that
returns something. We currently instantiate this such that it tries a
trusted gateway first (for performance reasons). However, eventually, it
would be nice if we tried chunks first since chunks are the easiest data
source to verify.

src/app.ts

@@ -25,10 +25,11 @@ import swaggerUi from 'swagger-ui-express';
 import YAML from 'yaml';
 
 import { ArweaveCompositeClient } from './arweave/composite-client.js';
-//import { ReadThroughChunkDataCache } from './cache/read-through-chunk-data-cache.js';
+import { ReadThroughChunkDataCache } from './cache/read-through-chunk-data-cache.js';
 import { GatewayDataSource } from './data/gateway-data-source.js';
+import { SequentialDataSource } from './data/sequential-data-source.js';
 import { StreamingManifestPathResolver } from './data/streaming-manifest-path-resolver.js';
-//import { TxChunksDataSource } from './data/tx-chunks-data-source.js';
+import { TxChunksDataSource } from './data/tx-chunks-data-source.js';
 import { StandaloneSqliteDatabase } from './database/standalone-sqlite.js';
 import { UniformFailureSimulator } from './lib/chaos.js';
 import log from './log.js';
@@ -40,7 +41,7 @@ import {
 } from './routes/data.js';
 import { apolloServer } from './routes/graphql/index.js';
 import { FsBlockStore } from './store/fs-block-store.js';
-//import { FsChunkDataStore } from './store/fs-chunk-data-store.js';
+import { FsChunkDataStore } from './store/fs-chunk-data-store.js';
 import { FsTransactionStore } from './store/fs-transaction-store.js';
 import { BlockImporter } from './workers/block-importer.js';
 import { TransactionFetcher } from './workers/transaction-fetcher.js';
@@ -138,23 +139,28 @@ const txRepairWorker = new TransactionRepairWorker({
 });
 
 // Configure contigous data source
-//const chunkDataSource = new ReadThroughChunkDataCache({
-//  log,
-//  chunkSource: arweaveClient,
-//  chunkDataStore: new FsChunkDataStore({ log, baseDir: 'data/chunks' }),
-//});
-//
-//const contiguousDataSource = new TxChunksDataSource({
-//  log,
-//  chainSource: arweaveClient,
-//  chunkSource: chunkDataSource,
-//});
-
-const contiguousDataSource = new GatewayDataSource({
+const chunkDataSource = new ReadThroughChunkDataCache({
+  log,
+  chunkSource: arweaveClient,
+  chunkDataStore: new FsChunkDataStore({ log, baseDir: 'data/chunks' }),
+});
+
+const txChunksDataSource = new TxChunksDataSource({
+  log,
+  chainSource: arweaveClient,
+  chunkSource: chunkDataSource,
+});
+
+const gatewayDataSource = new GatewayDataSource({
   log,
   trustedGatewayUrl,
 });
 
+const contiguousDataSource = new SequentialDataSource({
+  log,
+  dataSources: [gatewayDataSource, txChunksDataSource, arweaveClient],
+});
+
 const manifestPathResolver = new StreamingManifestPathResolver({
   log,
 });

src/data/sequential-data-source.ts

@@ -0,0 +1,38 @@
+import winston from 'winston';
+
+import { ContiguousData, ContiguousDataSource } from '../types.js';
+
+export class SequentialDataSource implements ContiguousDataSource {
+  private log: winston.Logger;
+  private dataSources: ContiguousDataSource[];
+
+  constructor({
+    log,
+    dataSources,
+  }: {
+    log: winston.Logger;
+    dataSources: ContiguousDataSource[];
+  }) {
+    this.log = log.child({ class: 'SequentialDataSource' });
+    this.dataSources = dataSources;
+  }
+
+  async getData(txId: string): Promise<ContiguousData> {
+    this.log.debug('Fetching contiguous data from data sources', { txId });
+
+    for (const dataSource of this.dataSources) {
+      try {
+        const data = await dataSource.getData(txId);
+        return data;
+      } catch (error: any) {
+        this.log.debug('Error fetching contiguous data from data source', {
+          txId,
+          message: error.message,
+          stack: error.stack,
+        });
+      }
+    }
+
+    throw new Error('Unable to fetch contiguous data from any data source');
+  }
+}