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 { createHttpPricingProvider } from "../../api/src/modules/pricing/providers/http-pricing-provider.js";
import { processNextPersistentImageDerivativeJob } from "./image-derivative-worker.js";
import {
  processNextPersistentPricingRefreshJob,
  runPricingRefreshJob,
} from "./pricing-refresh-worker.js";

const pollIntervalMs = Number.parseInt(
  process.env.WORKER_POLL_INTERVAL_MS ?? "2000",
  10,
);

async function runWorker() {
  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 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 processOne() {
    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();
  const timer = setInterval(
    () => processOne().catch((error) => console.error(error)),
    pollIntervalMs,
  );

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

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