feat(release): update 72ae3db

.vscode/settings.json

@@ -24,6 +24,7 @@
  "INJECTEDAPTOS",
  "Irys",
  "knexfile",
+ "localstack",
  "MULTIAPTOS",
  "multistream",
  "NOWAIT",

ecs/fulfillment-pipeline/src/app.ts

@@ -17,6 +17,7 @@
 import { Message } from "@aws-sdk/client-sqs";
 import { Consumer } from "sqs-consumer";
 
+import { ArweaveGateway } from "../../../src/arch/arweaveGateway";
 import { PostgresDatabase } from "../../../src/arch/db/postgres";
 import { TurboPaymentService } from "../../../src/arch/payment";
 import { migrateOnStartup } from "../../../src/constants";
@@ -73,6 +74,7 @@ const uploadDatabase = new PostgresDatabase({
 });
 const objectStore = getS3ObjectStore();
 const paymentService = new TurboPaymentService();
+const arweaveGateway = new ArweaveGateway();
 
 // Set up queue handler configurations for jobs based on a planId
 export const queues: QueueHandlerConfig[] = [
@@ -115,6 +117,7 @@ const consumers: ConsumerQueue[] = queues.map((queue) => ({
  database: uploadDatabase,
  objectStore,
  paymentService,
+ arweaveGateway,
  }),
  ...queue,
 }));
@@ -306,13 +309,15 @@ if (process.env.PLAN_BUNDLE_ENABLED === "true") {
  planBundleJobScheduler = new PlanBundleJobScheduler({
  intervalMs: +(process.env.PLAN_BUNDLE_INTERVAL_MS ?? 60_000),
  logger: globalLogger,
+ database: uploadDatabase,
  });
  setUpAndStartJobScheduler(planBundleJobScheduler);
 }
 if (process.env.VERIFY_BUNDLE_ENABLED === "true") {
  verifyBundleJobScheduler = new VerifyBundleJobScheduler({
  intervalMs: +(process.env.VERIFY_BUNDLE_INTERVAL_MS ?? 60_000),
  logger: globalLogger,
+ database: uploadDatabase,
  });
  setUpAndStartJobScheduler(verifyBundleJobScheduler);
 }

ecs/fulfillment-pipeline/src/jobs/plan.ts

@@ -16,22 +16,28 @@
  */
 import winston from "winston";
 
+import { Database } from "../../../../src/arch/db/database";
 import { planBundleHandler } from "../../../../src/jobs/plan";
 import { JobScheduler } from "../utils/jobScheduler";
 
 export class PlanBundleJobScheduler extends JobScheduler {
+ private database: Database;
+
  constructor({
  intervalMs = 60_000,
  logger,
+ database,
  }: {
  intervalMs: number;
  logger: winston.Logger;
+ database: Database;
  }) {
  super({ intervalMs, schedulerName: "plan-bundle", logger });
+ this.database = database;
  }
 
  async processJob(): Promise<void> {
- await planBundleHandler(undefined, undefined, this.logger).catch(
+ await planBundleHandler(this.database, undefined, this.logger).catch(
  (error) => {
  this.logger.error("Error planning bundle", error);
  }

ecs/fulfillment-pipeline/src/jobs/verify.ts

@@ -16,26 +16,34 @@
  */
 import winston from "winston";
 
+import { Database } from "../../../../src/arch/db/database";
 import { verifyBundleHandler } from "../../../../src/jobs/verify";
 import { JobScheduler } from "../utils/jobScheduler";
 
 export class VerifyBundleJobScheduler extends JobScheduler {
+ private database: Database;
  constructor({
  intervalMs = 60_000,
  logger,
+ database,
  }: {
  intervalMs: number;
  logger: winston.Logger;
+ database: Database;
  }) {
  super({
  intervalMs,
  schedulerName: "verify-bundle",
  logger,
  });
+ this.database = database;
  }
 
  async processJob(): Promise<void> {
- await verifyBundleHandler({ logger: this.logger }).catch((error) => {
+ await verifyBundleHandler({
+ database: this.database,
+ logger: this.logger,
+ }).catch((error) => {
  this.logger.error("Error verifying bundle", error);
  });
  }

ecs/fulfillment-pipeline/src/utils/planIdMessageHandler.ts

@@ -18,6 +18,7 @@ import { Message, SQSClient, SQSClientConfig } from "@aws-sdk/client-sqs";
 import { Consumer } from "sqs-consumer";
 import winston from "winston";
 
+import { ArweaveGateway } from "../../../../src/arch/arweaveGateway";
 import { Database } from "../../../../src/arch/db/database";
 import { ObjectStore } from "../../../../src/arch/objectStore";
 import { PaymentService } from "../../../../src/arch/payment";
@@ -31,13 +32,15 @@ export const planIdMessageHandler = ({
  database,
  objectStore,
  paymentService,
+ arweaveGateway,
 }: {
  message: Message;
  logger: winston.Logger;
  queue: QueueHandlerConfig;
  database: Database;
  objectStore: ObjectStore;
  paymentService: PaymentService;
+ arweaveGateway: ArweaveGateway;
 }) => {
  const messageLogger = logger.child({
  messageId: message.MessageId,
@@ -68,6 +71,7 @@ export const planIdMessageHandler = ({
  database,
  objectStore,
  paymentService,
+ arweaveGateway,
  },
  // provide our message logger to the handler
  messageLogger.child({ planId })
@@ -80,12 +84,14 @@ export function createPlanIdHandlingSQSConsumer({
  database,
  objectStore,
  paymentService,
+ arweaveGateway,
 }: {
  queue: QueueHandlerConfig;
  sqsOptions?: Partial<SQSClientConfig>;
  database: Database;
  objectStore: ObjectStore;
  paymentService: PaymentService;
+ arweaveGateway: ArweaveGateway;
 }) {
  const { queueUrl, consumerOptions, logger } = queue;
  return Consumer.create({
@@ -98,6 +104,7 @@ export function createPlanIdHandlingSQSConsumer({
  database,
  objectStore,
  paymentService,
+ arweaveGateway,
  }),
  sqs: new SQSClient(sqsOptions),
  batchSize: 1,

src/arch/arweaveGateway.ts

@@ -80,7 +80,7 @@ export class ArweaveGateway implements Gateway {
  endpoint = gatewayUrl,
  retryStrategy = new ExponentialBackoffRetryStrategy({}),
  axiosInstance = axios.create(), // defaults to throwing errors for status codes >400
- }: GatewayAPIConstParams) {
+ }: GatewayAPIConstParams = {}) {
  this.endpoint = endpoint;
  this.retryStrategy = retryStrategy;
  this.axiosInstance = axiosInstance;

src/arch/db/database.ts

@@ -15,6 +15,7 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 import {
+ DataItemFailedReason,
  FinishedMultiPartUpload,
  InFlightMultiPartUpload,
  InsertNewBundleParams,
@@ -27,7 +28,12 @@ import {
  PostedNewDataItem,
  SeededBundle,
 } from "../../types/dbTypes";
-import { TransactionId, UploadId, Winston } from "../../types/types";
+import {
+ DataItemId,
+ TransactionId,
+ UploadId,
+ Winston,
+} from "../../types/types";
 
 // TODO: this could be an interface since no functions have a default implementation
 export interface Database {
@@ -125,11 +131,12 @@ export interface Database {
  /** Gets latest status of a data item from the database */
  getDataItemInfo(dataItemId: TransactionId): Promise<
  | {
- status: "new" | "pending" | "permanent";
+ status: "new" | "pending" | "permanent" | "failed";
  assessedWinstonPrice: Winston;
  bundleId?: TransactionId;
  uploadedTimestamp: number;
  deadlineHeight?: number;
+ failedReason?: DataItemFailedReason;
  }
  | undefined
  >;
@@ -176,8 +183,10 @@ export interface Database {
  uploadId: UploadId
  ): Promise<void>;
 
- /** TODO: create failed_data_item table instead, remove this */
- deletePlannedDataItem(dataItemId: string): Promise<void>;
+ updatePlannedDataItemAsFailed(params: {
+ dataItemId: DataItemId;
+ failedReason: DataItemFailedReason;
+ }): Promise<void>;
 }
 
 export type UpdateDataItemsToPermanentParams = {

src/arch/db/dbConstants.ts

@@ -22,6 +22,7 @@ export const tableNames = {
  permanentBundle: "permanent_bundle",
  permanentDataItem: "permanent_data_item",
  plannedDataItem: "planned_data_item",
+ failedDataItem: "failed_data_item",
  postedBundle: "posted_bundle",
  seededBundle: "seeded_bundle",
  /** @deprecated */

src/arch/db/dbMaps.ts

@@ -16,8 +16,11 @@
  */
 import { defaultPremiumFeatureType } from "../../constants";
 import {
+ DataItemFailedReason,
  FailedBundle,
  FailedBundleDBResult,
+ FailedDataItem,
+ FailedDataItemDBResult,
  NewBundle,
  NewBundleDBResult,
  NewDataItem,
@@ -173,3 +176,41 @@ export function permanentDataItemDbResultToPermanentDataItemMap({
  deadlineHeight: deadline_height ? +deadline_height : undefined,
  };
 }
+
+export function failedDataItemDbResultToFailedDataItemMap({
+ assessed_winston_price,
+ byte_count,
+ data_item_id,
+ owner_public_address,
+ uploaded_date,
+ data_start,
+ failed_bundles,
+ signature_type,
+ content_type,
+ premium_feature_type,
+ plan_id,
+ planned_date,
+ deadline_height,
+ failed_date,
+ failed_reason,
+ signature,
+}: FailedDataItemDBResult): FailedDataItem {
+ return {
+ assessedWinstonPrice: W(assessed_winston_price),
+ dataItemId: data_item_id,
+ ownerPublicAddress: owner_public_address,
+ byteCount: +byte_count,
+ uploadedDate: uploaded_date,
+ premiumFeatureType: premium_feature_type ?? defaultPremiumFeatureType,
+ failedBundles: failed_bundles ? failed_bundles.split(",") : [],
+ signatureType: signature_type ?? undefined,
+ payloadDataStart: data_start ?? undefined,
+ payloadContentType: content_type ?? undefined,
+ signature: signature ?? undefined,
+ planId: plan_id,
+ plannedDate: planned_date,
+ deadlineHeight: deadline_height ? +deadline_height : undefined,
+ failedDate: failed_date,
+ failedReason: failed_reason as DataItemFailedReason,
+ };
+}