pokemon-inventory/apps/worker/src/worker-runtime.js

import dotenv from "dotenv";
import { fileURLToPath } from "node:url";

dotenv.config({
  path: fileURLToPath(new URL("../../../.env", import.meta.url)),
});

import { createDatabase } from "../../api/src/infrastructure/database/client.js";
import {
  createLocalImageStorage,
  resolveProjectImageStorageRoot,
} from "../../api/src/infrastructure/images/local-image-storage.js";
import { createImageDerivativeProcessor } from "../../api/src/infrastructure/images/image-derivative-processor.js";
import { createSharpImageRenderer } from "../../api/src/infrastructure/images/sharp-image-renderer.js";
import { createImageDerivativeJobRepository } from "../../api/src/modules/inventory/repositories/image-derivative-job-repository.js";
import { createPricingRepository } from "../../api/src/modules/pricing/repositories/pricing-repository.js";
import { createPricingRefreshJobRepository } from "../../api/src/modules/pricing/repositories/pricing-refresh-job-repository.js";
import { createPricingService } from "../../api/src/modules/pricing/services/pricing-service.js";
import { createPricingProviderRegistry } from "../../api/src/modules/pricing/providers/pricing-provider.js";
import { createFixturePricingProvider } from "../../api/src/modules/pricing/providers/fixture-pricing-provider.js";
import {
  assertProductionPricingProviderConfig,
  createHttpPricingProvider,
} from "../../api/src/modules/pricing/providers/http-pricing-provider.js";
import { createInventoryItemRepository } from "../../api/src/modules/inventory/repositories/inventory-item-repository.js";
import { createInventoryItemService } from "../../api/src/modules/inventory/services/inventory-item-service.js";
import { createInventoryExportJobRepository } from "../../api/src/modules/exports/repositories/inventory-export-job-repository.js";
import { createAdminExportService } from "../../api/src/modules/exports/services/admin-export-service.js";
import { processNextInventoryExportJob } from "./inventory-export-worker.js";
import { processNextPersistentImageDerivativeJob } from "./image-derivative-worker.js";
import {
  processNextPersistentPricingRefreshJob,
  purgeCompletedPricingRefreshJobs,
  runPricingRefreshJob,
} from "./pricing-refresh-worker.js";

const pollIntervalMs = Number.parseInt(
  process.env.WORKER_POLL_INTERVAL_MS ?? "2000",
  10,
);
const pricingJobRetentionDays = Number.parseInt(
  process.env.PRICING_JOB_RETENTION_DAYS ?? "30",
  10,
);
const pricingJobCleanupIntervalMs = Number.parseInt(
  process.env.PRICING_JOB_CLEANUP_INTERVAL_MS ?? "3600000",
  10,
);

async function runWorker() {
  if (process.env.NODE_ENV === "production") {
    assertProductionPricingProviderConfig({
      fixtureEnabled: process.env.PRICING_FIXTURE_PROVIDER_ENABLED === "true",
      endpoint: process.env.PRICING_HTTP_PROVIDER_ENDPOINT,
    });
  }
  const imageStorageRoot = resolveProjectImageStorageRoot(
    process.env.IMAGE_STORAGE_ROOT ?? ".local/image-storage",
    import.meta.url,
  );
  const database = createDatabase();
  const imageStorage = createLocalImageStorage(imageStorageRoot);
  const jobRepository = createImageDerivativeJobRepository(database.db);
  const pricingRefreshJobRepository = createPricingRefreshJobRepository(
    database.db,
  );
  const inventoryExportJobRepository = createInventoryExportJobRepository(database.db);
  const adminExportService = createAdminExportService({
    inventoryExportJobRepository,
    inventoryItemService: createInventoryItemService({
      inventoryItemRepository: createInventoryItemRepository(database.db),
    }),
  });
  const providers = [];
  if (process.env.PRICING_FIXTURE_PROVIDER_ENABLED === "true")
    providers.push(createFixturePricingProvider());
  if (process.env.PRICING_HTTP_PROVIDER_ENDPOINT) {
    providers.push(
      createHttpPricingProvider({
        id: process.env.PRICING_HTTP_PROVIDER_ID ?? "http-provider",
        endpoint: process.env.PRICING_HTTP_PROVIDER_ENDPOINT,
        token: process.env.PRICING_HTTP_PROVIDER_TOKEN,
      }),
    );
  }
  const pricingProviderRegistry = createPricingProviderRegistry(providers);
  const pricingService = createPricingService({
    pricingRepository: createPricingRepository(database.db),
    pricingProviderRegistry,
  });
  const processor = createImageDerivativeProcessor({
    imageStorage,
    renderer: createSharpImageRenderer(),
  });

  async function cleanupPricingJobs() {
    const deletedCount = await purgeCompletedPricingRefreshJobs({
      jobRepository: pricingRefreshJobRepository,
      retentionDays: pricingJobRetentionDays,
    });
    if (deletedCount > 0)
      console.log(`Purged ${deletedCount} completed pricing refresh job(s).`);
  }

  async function processOne() {
    const exportResult = await processNextInventoryExportJob({
      jobRepository: inventoryExportJobRepository,
      exportInventory: (job) => adminExportService.inventoryJson({ sellerId: job.sellerId }),
    });
    if (exportResult)
      console.log(`Inventory export job ${exportResult.id} ${exportResult.status.toLowerCase()}.`);

    const result = await processNextPersistentImageDerivativeJob({
      jobRepository,
      async loadSource(sourceKey) {
        return { sourceKey, data: await imageStorage.get(sourceKey) };
      },
      processor,
    });

    if (result)
      console.log(
        `Image derivative job ${result.id} ${result.status.toLowerCase()}.`,
      );

    const pricingResult = await processNextPersistentPricingRefreshJob({
      jobRepository: pricingRefreshJobRepository,
      pricingRefreshJob: (job) =>
        runPricingRefreshJob({
          job,
          providerRegistry: pricingProviderRegistry,
          pricingService,
        }),
    });

    if (pricingResult)
      console.log(
        `Pricing refresh job ${pricingResult.id} ${pricingResult.status.toLowerCase()}.`,
      );
  }

  await processOne();
  await cleanupPricingJobs();
  const timer = setInterval(
    () => processOne().catch((error) => console.error(error)),
    pollIntervalMs,
  );
  const cleanupTimer = setInterval(
    () => cleanupPricingJobs().catch((error) => console.error(error)),
    pricingJobCleanupIntervalMs,
  );

  const shutdown = async () => {
    clearInterval(timer);
    clearInterval(cleanupTimer);
    await database.close();
  };
  process.once("SIGINT", shutdown);
  process.once("SIGTERM", shutdown);
}

runWorker().catch((error) => {
  console.error(error);
  process.exitCode = 1;
});