feat: reconcile application vertical slices
Backfill the completed acquisitions, sales, dashboard, pricing, search, public detail, worker, migration, contracts, validation, tests, and implementation-status changes into one verified application slice.
This commit is contained in:
@@ -1,25 +1,40 @@
|
||||
export async function processNextPersistentPricingRefreshJob({ jobRepository, pricingRefreshJob }) {
|
||||
export async function processNextPersistentPricingRefreshJob({
|
||||
jobRepository,
|
||||
pricingRefreshJob,
|
||||
}) {
|
||||
const claimed = await jobRepository.claimNext();
|
||||
if (!claimed) return null;
|
||||
|
||||
const result = await pricingRefreshJob({ ...claimed, type: 'PRICING_REFRESH' });
|
||||
if (result.status === 'COMPLETED') {
|
||||
await jobRepository.complete(claimed.id, result.observations, result.valuation);
|
||||
const result = await pricingRefreshJob({
|
||||
...claimed,
|
||||
type: "PRICING_REFRESH",
|
||||
});
|
||||
if (result.status === "COMPLETED") {
|
||||
await jobRepository.complete(
|
||||
claimed.id,
|
||||
result.observations,
|
||||
result.valuation,
|
||||
);
|
||||
} else {
|
||||
await jobRepository.fail(claimed.id, result.error);
|
||||
}
|
||||
return result;
|
||||
}export async function runPricingRefreshJob({ job, providerRegistry, pricingService }) {
|
||||
if (!job || job.type !== 'PRICING_REFRESH') {
|
||||
throw new Error('Unsupported pricing refresh job.');
|
||||
}
|
||||
export async function runPricingRefreshJob({
|
||||
job,
|
||||
providerRegistry,
|
||||
pricingService,
|
||||
}) {
|
||||
if (!job || job.type !== "PRICING_REFRESH") {
|
||||
throw new Error("Unsupported pricing refresh job.");
|
||||
}
|
||||
|
||||
const provider = providerRegistry?.get(job.providerId);
|
||||
if (!provider) {
|
||||
return {
|
||||
...job,
|
||||
status: 'FAILED',
|
||||
error: 'Pricing provider is not configured.',
|
||||
status: "FAILED",
|
||||
error: "Pricing provider is not configured.",
|
||||
failedAt: new Date(),
|
||||
};
|
||||
}
|
||||
@@ -29,17 +44,21 @@ export async function processNextPersistentPricingRefreshJob({ jobRepository, pr
|
||||
inventoryItemId: job.inventoryItemId,
|
||||
productId: job.productId,
|
||||
conditionId: job.conditionId,
|
||||
externalIdentifier: job.externalIdentifier,
|
||||
searchQuery: job.searchQuery,
|
||||
});
|
||||
const savedObservations = [];
|
||||
for (const observation of observations) {
|
||||
savedObservations.push(await pricingService.addObservation({
|
||||
sellerId: job.sellerId,
|
||||
...observation,
|
||||
providerId: provider.id,
|
||||
inventoryItemId: job.inventoryItemId,
|
||||
productId: job.productId,
|
||||
conditionId: job.conditionId,
|
||||
}));
|
||||
savedObservations.push(
|
||||
await pricingService.addObservation({
|
||||
sellerId: job.sellerId,
|
||||
...observation,
|
||||
providerId: provider.id,
|
||||
inventoryItemId: job.inventoryItemId,
|
||||
productId: job.productId,
|
||||
conditionId: job.conditionId,
|
||||
}),
|
||||
);
|
||||
}
|
||||
const valuation = await pricingService.calculate({
|
||||
sellerId: job.sellerId,
|
||||
@@ -48,7 +67,7 @@ export async function processNextPersistentPricingRefreshJob({ jobRepository, pr
|
||||
|
||||
return {
|
||||
...job,
|
||||
status: 'COMPLETED',
|
||||
status: "COMPLETED",
|
||||
observations: savedObservations,
|
||||
valuation,
|
||||
completedAt: new Date(),
|
||||
@@ -56,8 +75,8 @@ export async function processNextPersistentPricingRefreshJob({ jobRepository, pr
|
||||
} catch (error) {
|
||||
return {
|
||||
...job,
|
||||
status: 'FAILED',
|
||||
error: error instanceof Error ? error.message : 'Pricing refresh failed.',
|
||||
status: "FAILED",
|
||||
error: error instanceof Error ? error.message : "Pricing refresh failed.",
|
||||
failedAt: new Date(),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -1,32 +1,59 @@
|
||||
import dotenv from 'dotenv';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
import dotenv from "dotenv";
|
||||
import { fileURLToPath } from "node:url";
|
||||
|
||||
dotenv.config({ path: fileURLToPath(new URL('../../../.env', import.meta.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 { processNextPersistentImageDerivativeJob } from './image-derivative-worker.js';
|
||||
import { processNextPersistentPricingRefreshJob, runPricingRefreshJob } from './pricing-refresh-worker.js';
|
||||
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);
|
||||
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 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 pricingProviderRegistry = process.env.PRICING_FIXTURE_PROVIDER_ENABLED === 'true'
|
||||
? createPricingProviderRegistry([createFixturePricingProvider()])
|
||||
: createPricingProviderRegistry();
|
||||
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,
|
||||
@@ -45,32 +72,42 @@ async function runWorker() {
|
||||
processor,
|
||||
});
|
||||
|
||||
if (result) console.log(`Image derivative job ${result.id} ${result.status.toLowerCase()}.`);
|
||||
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,
|
||||
}),
|
||||
pricingRefreshJob: (job) =>
|
||||
runPricingRefreshJob({
|
||||
job,
|
||||
providerRegistry: pricingProviderRegistry,
|
||||
pricingService,
|
||||
}),
|
||||
});
|
||||
|
||||
if (pricingResult) console.log(`Pricing refresh job ${pricingResult.id} ${pricingResult.status.toLowerCase()}.`);
|
||||
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 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);
|
||||
process.once("SIGINT", shutdown);
|
||||
process.once("SIGTERM", shutdown);
|
||||
}
|
||||
|
||||
runWorker().catch((error) => {
|
||||
console.error(error);
|
||||
process.exitCode = 1;
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,78 +1,111 @@
|
||||
import assert from 'node:assert/strict';
|
||||
import test from 'node:test';
|
||||
import assert from "node:assert/strict";
|
||||
import test from "node:test";
|
||||
|
||||
import { createPricingProvider, createPricingProviderRegistry } from '../../api/src/modules/pricing/providers/pricing-provider.js';
|
||||
import { processNextPersistentPricingRefreshJob, runPricingRefreshJob } from '../src/pricing-refresh-worker.js';
|
||||
import {
|
||||
createPricingProvider,
|
||||
createPricingProviderRegistry,
|
||||
} from "../../api/src/modules/pricing/providers/pricing-provider.js";
|
||||
import {
|
||||
processNextPersistentPricingRefreshJob,
|
||||
runPricingRefreshJob,
|
||||
} from "../src/pricing-refresh-worker.js";
|
||||
|
||||
test('completes a pricing refresh through the provider and pricing service boundaries', async () => {
|
||||
test("completes a pricing refresh through the provider and pricing service boundaries", async () => {
|
||||
const providerRegistry = createPricingProviderRegistry([
|
||||
createPricingProvider({
|
||||
id: 'fixture-provider',
|
||||
id: "fixture-provider",
|
||||
async observe() {
|
||||
return [{ priceMinor: 2500, sourceIdentifier: 'fixture-1' }];
|
||||
return [{ priceMinor: 2500, sourceIdentifier: "fixture-1" }];
|
||||
},
|
||||
}),
|
||||
]);
|
||||
const saved = [];
|
||||
const result = await runPricingRefreshJob({
|
||||
job: {
|
||||
id: 'job-1',
|
||||
type: 'PRICING_REFRESH',
|
||||
providerId: 'fixture-provider',
|
||||
sellerId: 'seller-1',
|
||||
productId: 'product-1',
|
||||
inventoryItemId: 'inventory-1',
|
||||
conditionId: 'condition-1',
|
||||
id: "job-1",
|
||||
type: "PRICING_REFRESH",
|
||||
providerId: "fixture-provider",
|
||||
sellerId: "seller-1",
|
||||
productId: "product-1",
|
||||
inventoryItemId: "inventory-1",
|
||||
conditionId: "condition-1",
|
||||
},
|
||||
providerRegistry,
|
||||
pricingService: {
|
||||
async addObservation(input) { saved.push(input); return input; },
|
||||
async calculate() { return { valuation: { valueMinor: 2500 } }; },
|
||||
async addObservation(input) {
|
||||
saved.push(input);
|
||||
return input;
|
||||
},
|
||||
async calculate() {
|
||||
return { valuation: { valueMinor: 2500 } };
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
assert.equal(result.status, 'COMPLETED');
|
||||
assert.equal(result.status, "COMPLETED");
|
||||
assert.equal(result.valuation.valuation.valueMinor, 2500);
|
||||
assert.equal(saved[0].providerId, 'fixture-provider');
|
||||
assert.equal(saved[0].providerId, "fixture-provider");
|
||||
});
|
||||
|
||||
test('fails a refresh clearly when a provider is unavailable', async () => {
|
||||
test("fails a refresh clearly when a provider is unavailable", async () => {
|
||||
const result = await runPricingRefreshJob({
|
||||
job: { id: 'job-2', type: 'PRICING_REFRESH', providerId: 'missing' },
|
||||
job: { id: "job-2", type: "PRICING_REFRESH", providerId: "missing" },
|
||||
providerRegistry: createPricingProviderRegistry(),
|
||||
pricingService: {},
|
||||
});
|
||||
|
||||
assert.equal(result.status, 'FAILED');
|
||||
assert.equal(result.status, "FAILED");
|
||||
assert.match(result.error, /not configured/);
|
||||
});
|
||||
|
||||
test('persists completed and failed outcomes through the job repository', async () => {
|
||||
test("persists completed and failed outcomes through the job repository", async () => {
|
||||
const calls = [];
|
||||
const jobRepository = {
|
||||
async claimNext() { return { id: 'job-3', type: 'PRICING_REFRESH', providerId: 'fixture' }; },
|
||||
async complete(id, observations, valuation) { calls.push(['complete', id, observations, valuation]); },
|
||||
async fail(id, error) { calls.push(['fail', id, error]); },
|
||||
async claimNext() {
|
||||
return { id: "job-3", type: "PRICING_REFRESH", providerId: "fixture" };
|
||||
},
|
||||
async complete(id, observations, valuation) {
|
||||
calls.push(["complete", id, observations, valuation]);
|
||||
},
|
||||
async fail(id, error) {
|
||||
calls.push(["fail", id, error]);
|
||||
},
|
||||
};
|
||||
|
||||
await processNextPersistentPricingRefreshJob({
|
||||
jobRepository,
|
||||
async pricingRefreshJob() { return { status: 'COMPLETED', observations: [{ id: 'observation-1' }], valuation: { valueMinor: 2500 } }; },
|
||||
async pricingRefreshJob() {
|
||||
return {
|
||||
status: "COMPLETED",
|
||||
observations: [{ id: "observation-1" }],
|
||||
valuation: { valueMinor: 2500 },
|
||||
};
|
||||
},
|
||||
});
|
||||
|
||||
assert.deepEqual(calls, [['complete', 'job-3', [{ id: 'observation-1' }], { valueMinor: 2500 }]]);
|
||||
assert.deepEqual(calls, [
|
||||
["complete", "job-3", [{ id: "observation-1" }], { valueMinor: 2500 }],
|
||||
]);
|
||||
});
|
||||
|
||||
test('persists failed pricing refresh outcomes through the repository', async () => {
|
||||
test("persists failed pricing refresh outcomes through the repository", async () => {
|
||||
const calls = [];
|
||||
await processNextPersistentPricingRefreshJob({
|
||||
jobRepository: {
|
||||
async claimNext() { return { id: 'job-4' }; },
|
||||
async complete() { throw new Error('not expected'); },
|
||||
async fail(id, error) { calls.push([id, error]); },
|
||||
async claimNext() {
|
||||
return { id: "job-4" };
|
||||
},
|
||||
async complete() {
|
||||
throw new Error("not expected");
|
||||
},
|
||||
async fail(id, error) {
|
||||
calls.push([id, error]);
|
||||
},
|
||||
},
|
||||
async pricingRefreshJob() {
|
||||
return { status: "FAILED", error: "Provider unavailable." };
|
||||
},
|
||||
async pricingRefreshJob() { return { status: 'FAILED', error: 'Provider unavailable.' }; },
|
||||
});
|
||||
|
||||
assert.deepEqual(calls, [['job-4', 'Provider unavailable.']]);
|
||||
assert.deepEqual(calls, [["job-4", "Provider unavailable."]]);
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user