Job Feed Change Detector
Pricing
from $30.00 / 1,000 job snapshots compareds
Compare baseline and current job-feed exports to find new, changed, removed, and duplicate postings with stable keys and field-level evidence.
Pricing
from $30.00 / 1,000 job snapshots compareds
Rating
0.0
(0)
Developer
Cedric Günther
Maintained by CommunityActor stats
0
Bookmarked
2
Total users
1
Monthly active users
4 days ago
Last modified
Categories
Share
{ "$schema": "https://apify.com/schemas/v1/actor.json", "actorSpecification": 1, "name": "job-feed-change-detector", "title": "Job Feed Change Detector", "description": "Compare baseline and current job-feed exports to find new, changed, removed, and duplicate postings with stable keys and field-level evidence.", "version": "0.1", "buildTag": "latest", "defaultMemoryMbytes": 256, "inputSchema": "./input_schema.json", "outputSchema": "./output_schema.json", "storages": { "dataset": "./dataset_schema.json" }, "dockerfile": "../Dockerfile"}{ "$schema": "https://apify.com/schemas/v1/dataset.ide.json", "actorSpecification": 1, "fields": { "$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "properties": { "recordType": { "type": "string" }, "comparisonId": { "type": "string" }, "stableKey": { "type": [ "string", "null" ] }, "changeType": { "type": [ "string", "null" ] }, "changedFields": { "type": [ "array", "null" ] }, "snapshot": { "type": [ "string", "null" ] }, "indexes": { "type": [ "array", "null" ], "items": { "type": "integer" } }, "count": { "type": [ "integer", "null" ] }, "engineVersion": { "type": "string" } }, "required": [ "recordType", "comparisonId", "engineVersion" ], "additionalProperties": true }, "views": { "changes": { "title": "Job Feed Change Detector", "transformation": { "fields": [ "recordType", "comparisonId", "stableKey", "changeType", "changedFields", "snapshot", "indexes", "count", "engineVersion" ] }, "display": { "component": "table", "properties": {} } } }}{ "$schema": "https://apify.com/schemas/v1/input_schema.json", "schemaVersion": 1, "title": "Job Feed Change Detector input", "type": "object", "required": [ "comparisonId", "baseline", "current" ], "additionalProperties": false, "properties": { "comparisonId": { "title": "Comparison ID", "type": "string", "minLength": 1, "maxLength": 120, "prefill": "sample-jobs", "editor": "textfield", "description": "Input field comparisonId." }, "baseline": { "title": "Baseline job records", "type": "array", "maxItems": 1000, "editor": "json", "prefill": [ { "sourceId": "J-1", "title": "Data Analyst", "company": "Northwind", "location": "Berlin", "url": "https://jobs.example/J-1", "description": "SQL and dashboards" } ], "items": { "type": "object", "required": [ "title", "company" ], "additionalProperties": false, "properties": { "sourceId": { "type": "string", "maxLength": 1000, "title": "Source Id", "description": "Input field sourceId." }, "title": { "type": "string", "minLength": 1, "maxLength": 500, "title": "Title", "description": "Input field title." }, "company": { "type": "string", "minLength": 1, "maxLength": 500, "title": "Company", "description": "Input field company." }, "location": { "type": "string", "maxLength": 1000, "title": "Location", "description": "Input field location." }, "url": { "type": "string", "maxLength": 2000, "title": "Url", "description": "Input field url." }, "description": { "type": "string", "maxLength": 20000, "title": "Description", "description": "Input field description." }, "employmentType": { "type": "string", "maxLength": 500, "title": "Employment Type", "description": "Input field employmentType." }, "salary": { "type": "string", "maxLength": 1000, "title": "Salary", "description": "Input field salary." } } }, "description": "Input field baseline." }, "current": { "title": "Current job records", "type": "array", "maxItems": 1000, "editor": "json", "prefill": [ { "sourceId": "J-1", "title": "Senior Data Analyst", "company": "Northwind", "location": "Berlin", "url": "https://jobs.example/J-1", "description": "SQL, dashboards, and Python" }, { "sourceId": "J-2", "title": "Data Engineer", "company": "Northwind", "location": "Remote", "url": "https://jobs.example/J-2", "description": "Build data pipelines" } ], "items": { "type": "object", "required": [ "title", "company" ], "additionalProperties": false, "properties": { "sourceId": { "type": "string", "maxLength": 1000, "title": "Source Id", "description": "Input field sourceId." }, "title": { "type": "string", "minLength": 1, "maxLength": 500, "title": "Title", "description": "Input field title." }, "company": { "type": "string", "minLength": 1, "maxLength": 500, "title": "Company", "description": "Input field company." }, "location": { "type": "string", "maxLength": 1000, "title": "Location", "description": "Input field location." }, "url": { "type": "string", "maxLength": 2000, "title": "Url", "description": "Input field url." }, "description": { "type": "string", "maxLength": 20000, "title": "Description", "description": "Input field description." }, "employmentType": { "type": "string", "maxLength": 500, "title": "Employment Type", "description": "Input field employmentType." }, "salary": { "type": "string", "maxLength": 1000, "title": "Salary", "description": "Input field salary." } } }, "description": "Input field current." }, "trackedFields": { "title": "Fields to compare", "type": "array", "minItems": 1, "maxItems": 30, "uniqueItems": true, "editor": "json", "items": { "type": "string", "minLength": 1, "maxLength": 100 }, "description": "Input field trackedFields." }, "maxRecordsPerSnapshot": { "title": "Maximum records per snapshot", "type": "integer", "minimum": 1, "maximum": 1000, "description": "Input field maxRecordsPerSnapshot." } }}{ "$schema": "https://apify.com/schemas/v1/output.ide.json", "actorOutputSchemaVersion": 1, "title": "Job Feed Change Detector output", "properties": { "evidence": { "type": "string", "title": "Dataset evidence", "template": "{{links.apiDefaultDatasetUrl}}/items" }, "summary": { "type": "string", "title": "Run summary", "template": "{{links.apiDefaultKeyValueStoreUrl}}/records/OUTPUT" }, "reports": { "type": "string", "title": "Complete report", "template": "{{links.apiDefaultKeyValueStoreUrl}}/records/REPORTS" } }}node_modulesdiststoragecoverage.envnode_modulesdiststoragecoverage.env{ "manifestSchemaVersion": "1.0.0", "identity": { "key": "job-feed-change-detector", "actorId": null, "slug": "job-feed-change-detector", "title": "Job Feed Change Detector", "description": "Compare baseline and current job-feed exports to find new, changed, removed, and duplicate postings with stable keys and field-level evidence.", "maintenanceClass": "A" }, "scope": { "supportedCapabilities": [ "job-feed-snapshot-comparison", "job-posting-deduplication", "field-level-job-change-evidence" ], "unsupportedCapabilities": [ "job-board-scraping", "applicant-ranking", "salary-estimation", "remote-dataset-access" ], "claimsBoundaries": [ "Version 1 compares bounded inline job records and never accesses a job board.", "Stable keys prefer source IDs or canonical URLs and otherwise use normalized company-title-location fields.", "Removed means absent from the supplied current snapshot, not necessarily closed at the source." ], "externalDependencies": [], "statefulness": "Run-local deterministic comparison; only the PPE ledger persists semantic comparison keys for retry-safe charging." }, "runtime": { "language": "JavaScript", "runtime": "Node.js >=22", "entryPoint": "src/main.js", "defaultMemoryMbytes": 256 }, "ppe": { "pricingModel": "PAY_PER_EVENT", "primaryEvent": "job-snapshots-compared", "events": [ { "name": "job-snapshots-compared", "title": "Job snapshots compared", "description": "One baseline/current job snapshot pair converted into durable change evidence.", "priceUsd": 0.03, "primary": true, "synthetic": false }, { "name": "apify-actor-start", "title": "Actor start", "description": "Platform-managed Actor start event.", "priceUsd": 5e-05, "primary": false, "synthetic": true } ], "billableBoundary": "One valid snapshot pair after the complete change report is durable.", "nonBillableFailureClasses": [ "invalid-actor-input", "resource-limit-exceeded", "internal-error" ], "platformUsagePolicy": "INCLUDED_IN_EVENT_PRICE", "durableBeforeCharge": true, "reconciliation": "chargedCount equals one per distinct completed comparison and zero for rejected input.", "idempotency": "SHA-256 over comparison ID, normalized baseline, normalized current snapshot, and tracked fields.", "spendingLimitBehavior": "The comparison does not begin when custom-event capacity is unavailable." }, "store": { "seo": { "title": "Job Feed Change & Duplicate Detector", "description": "Compare job feed exports to detect new, changed, removed, and duplicate postings with stable keys and field-level evidence." }, "sourceVisibility": "VISIBLE", "permissions": [ "LIMITED_PERMISSIONS", "default-storage-read-write" ], "categories": [ "JOBS", "AUTOMATION" ], "storeQaPrefill": { "comparisonId": "sample-jobs", "baseline": [ { "sourceId": "J-1", "title": "Data Analyst", "company": "Northwind", "location": "Berlin", "url": "https://jobs.example/J-1", "description": "SQL and dashboards" } ], "current": [ { "sourceId": "J-1", "title": "Senior Data Analyst", "company": "Northwind", "location": "Berlin", "url": "https://jobs.example/J-1", "description": "SQL, dashboards, and Python" }, { "sourceId": "J-2", "title": "Data Engineer", "company": "Northwind", "location": "Remote", "url": "https://jobs.example/J-2", "description": "Build data pipelines" } ] }, "expectedStoreQaOutput": "A deterministic summary with one changed and one new posting plus field-level changes and stable keys.", "maxExpectedRuntimeSeconds": 30, "repeatabilityRuns": 2, "billingCeilingUsd": 0.03005 }, "tasks": [ { "key": "compare-job-feed-snapshots", "taskId": null, "slug": "compare-job-feed-snapshots", "title": "Compare two job feed snapshots", "description": "Identify new, changed, removed, and duplicate job postings between two safe inline exports.", "seo": { "title": "Compare Job Feed Snapshots for Changes", "description": "Find new, changed, removed, and duplicate job listings between two ATS or scraper exports with deterministic evidence.", "intent": "compare two job listing feed exports for additions removals changes and duplicates" }, "savedInput": { "comparisonId": "example-jobs", "baseline": [ { "sourceId": "A-1", "title": "Support Engineer", "company": "Example Co", "location": "London", "url": "https://jobs.example/A-1" } ], "current": [ { "sourceId": "A-1", "title": "Senior Support Engineer", "company": "Example Co", "location": "London", "url": "https://jobs.example/A-1" } ] }, "publicFields": [ "comparisonId", "baseline", "current" ], "hiddenFields": [ "trackedFields", "maxRecordsPerSnapshot" ], "datasetView": "changes", "scopeCaveat": "Inline records only; absence from the current input is not independently verified source closure.", "capabilityClaims": [ "job-feed-snapshot-comparison", "field-level-job-change-evidence" ], "safeFixture": true, "validationBudgetUsd": 0.03005, "status": "PLANNED" } ], "qa": { "localTestCommands": [ "npm test", "npm run build", "npm run schema:check", "npm run secret:scan" ], "ppeTestCommands": [ "npm run test -- --test-name-pattern=ppe" ], "storeQaStaticContract": "tools/store-qa-gate/store_qa_gate.py --schema-only actors/job-feed-change-detector", "local": { "status": "PASS", "reference": "actors/job-feed-change-detector/reports/LOCAL_QA.md" }, "ppeLocal": { "status": "PASS", "reference": "actors/job-feed-change-detector/reports/LOCAL_QA.md" }, "ppeLive": { "status": "NOT_RUN", "reference": null }, "storeQaCloudProof": { "status": "NOT_RUN", "reference": null }, "taskLive": { "status": "NOT_RUN", "reference": null } }, "maintenance": { "class": "A", "expectedMinutesPerMonth": "5-10", "knownFragilityClasses": [ "source-field-alias-drift", "canonical-url-edge-cases" ], "status": "OBSERVE" }, "release": { "validatedVersion": null, "validatedBuildId": null, "visibility": "UNDEPLOYED", "publicationStatus": "SCAFFOLD_ONLY", "lastValidationDate": null }}FROM apify/actor-node:24 AS builderCOPY package*.json ./RUN npm ci --include=dev --audit=falseCOPY . ./RUN npm run buildFROM apify/actor-node:24COPY package*.json ./RUN npm ci --omit=dev --omit=optional --audit=false && npm cache clean --forceCOPY /usr/src/app/dist ./distCOPY .actor ./.actorCMD ["node", "dist/src/main.js"]{ "name": "job-feed-change-detector", "version": "0.1.0", "private": true, "type": "module", "description": "Deterministic inline job feed snapshot comparison Actor.", "engines": { "node": ">=22" }, "scripts": { "start": "tsx src/main.ts", "start:prod": "node dist/src/main.js", "build": "tsc", "benchmark": "node scripts/benchmark.mjs", "test": "vitest run", "schema:check": "node dist/scripts/validate-schemas.js", "secret:scan": "node dist/scripts/secret-scan.js", "quality": "npm test && npm run build && npm run schema:check && npm run secret:scan" }, "dependencies": { "apify": "3.7.2" }, "devDependencies": { "@apify/tsconfig": "0.2.0", "@types/node": "24.3.0", "tsx": "4.20.5", "typescript": "5.9.2", "vitest": "3.2.4" }}1import { processInput } from '../dist/src/domain.js';2const baseline = Array.from({ length: 1000 }, (_, index) => ({ sourceId: `s${index}`, title: `Role ${index}`, company: 'Example' }));3const current = baseline.map((record, index) => ({ ...record, title: `Senior Role ${index}` }));4const input = { comparisonId: 'max-bound', baseline, current };5const before = process.memoryUsage().heapUsed;6const started = performance.now();7const result = processInput(input);8console.log(JSON.stringify({ actor: 'job-feed-change-detector', workload: '1000 plus 1000 records', wallMs: Number((performance.now() - started).toFixed(3)), heapDeltaBytes: process.memoryUsage().heapUsed - before, outputRecords: result.changes.length + result.duplicates.length }));1import { readdir, readFile } from 'node:fs/promises';2import { extname, join } from 'node:path';3
4const patterns = [5 /-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----/,6 /\b(?:AKIA|ASIA)[A-Z0-9]{16}\b/,7 /\b(?:apify_api_|sk-)[A-Za-z0-9_-]{20,}\b/,8];9let scanned = 0;10async function walk(directory: string): Promise<void> {11 for (const entry of await readdir(directory, { withFileTypes: true })) {12 const path = join(directory, entry.name);13 if (entry.isDirectory()) await walk(path);14 else if (['.ts', '.json', '.md'].includes(extname(path))) {15 scanned += 1;16 const content = await readFile(path, 'utf8');17 if (patterns.some((pattern) => pattern.test(content)))18 throw new Error(`Potential secret in ${path}`);19 }20 }21}22for (const root of ['src', 'test', 'scripts', '.actor']) await walk(root);23console.log(JSON.stringify({ status: 'PASS', scannedFiles: scanned }));1import { readFile } from 'node:fs/promises';2import { join } from 'node:path';3
4const files = ['actor.json', 'input_schema.json', 'output_schema.json', 'dataset_schema.json'];5for (const file of files) {6 const value = JSON.parse(7 await readFile(join(process.cwd(), '.actor', file), 'utf8'),8 ) as Record<string, unknown>;9 if (typeof value.$schema !== 'string') throw new Error(`${file}: missing $schema`);10}11const dataset = JSON.parse(12 await readFile(join(process.cwd(), '.actor', 'dataset_schema.json'), 'utf8'),13) as { views?: Record<string, unknown> };14if (!dataset.views?.changes) throw new Error('dataset_schema.json: changes view is required');15console.log(JSON.stringify({ status: 'PASS', schemas: files.length, datasetView: 'changes' }));1import { createHash } from 'node:crypto';2
3export const ENGINE_VERSION = '1.0.0';4const DEFAULT_TRACKED_FIELDS = ['title', 'company', 'location', 'url', 'description', 'employmentType', 'salary'];5export type JobRecord = Record<string, string | number | boolean | null | undefined> & {6 sourceId?: string;7 title: string;8 company: string;9 location?: string;10 url?: string;11};12export interface Change {13 recordType: 'job-change';14 comparisonId: string;15 stableKey: string;16 changeType: 'ADDED' | 'REMOVED' | 'CHANGED';17 changedFields: { field: string; before: unknown; after: unknown }[];18 baseline: JobRecord | null;19 current: JobRecord | null;20 engineVersion: string;21}22export interface DuplicateGroup {23 recordType: 'duplicate-group';24 comparisonId: string;25 snapshot: 'baseline' | 'current';26 stableKey: string;27 indexes: number[];28 count: number;29 engineVersion: string;30}31export interface RunResult {32 inputHash: string;33 changes: Change[];34 duplicates: DuplicateGroup[];35 summary: {36 recordType: 'job-summary';37 status: 'SUCCEEDED';38 comparisonId: string;39 baselineRecords: number;40 currentRecords: number;41 added: number;42 removed: number;43 changed: number;44 unchanged: number;45 duplicateGroups: number;46 inputHash: string;47 engineVersion: string;48 };49}50export class DomainError extends Error {51 public constructor(public readonly code: string, message: string) {52 super(message);53 }54}55function canonical(value: unknown): string {56 if (Array.isArray(value)) return `[${value.map(canonical).join(',')}]`;57 if (value && typeof value === 'object') {58 return `{${Object.entries(value as Record<string, unknown>)59 .sort(([a], [b]) => a.localeCompare(b))60 .map(([key, child]) => `${JSON.stringify(key)}:${canonical(child)}`)61 .join(',')}}`;62 }63 return JSON.stringify(value);64}65function hash(value: unknown): string {66 return createHash('sha256').update(canonical(value)).digest('hex');67}68function object(value: unknown, label: string): Record<string, unknown> {69 if (!value || typeof value !== 'object' || Array.isArray(value))70 throw new DomainError('INVALID_INPUT', `${label} must be an object`);71 return value as Record<string, unknown>;72}73function text(value: unknown): string {74 return typeof value === 'string'75 ? value.normalize('NFKC').toLocaleLowerCase('en-US').trim().replace(/\s+/g, ' ')76 : String(value ?? '');77}78function canonicalUrl(value: string): string {79 let url: URL;80 try {81 url = new URL(value);82 } catch {83 throw new DomainError('INVALID_INPUT', 'job url must be a valid absolute HTTP or HTTPS URL');84 }85 if (!['http:', 'https:'].includes(url.protocol))86 throw new DomainError('INVALID_INPUT', 'job url must use HTTP or HTTPS');87 url.hash = '';88 url.hostname = url.hostname.toLowerCase().replace(/^www\./, '');89 for (const key of [...url.searchParams.keys()]) {90 if (/^(utm_|gclid$|fbclid$|ref$)/i.test(key)) url.searchParams.delete(key);91 }92 url.searchParams.sort();93 url.pathname = url.pathname.replace(/\/+$/, '') || '/';94 return url.toString();95}96export function stableJobKey(record: JobRecord): string {97 if (record.sourceId && text(record.sourceId)) return `source:${text(record.sourceId)}`;98 if (record.url && text(record.url)) return `url:${canonicalUrl(record.url)}`;99 const fallback = [record.company, record.title, record.location].map(text);100 if (!fallback[0] || !fallback[1] || !fallback[2])101 throw new DomainError('INVALID_INPUT', 'job needs sourceId, url, or company/title/location');102 return `tuple:${fallback.join('|')}`;103}104function parseRows(value: unknown, label: string, maximum: number): JobRecord[] {105 if (!Array.isArray(value) || value.length > maximum)106 throw new DomainError('RESOURCE_LIMIT_EXCEEDED', `${label} must be an array with at most ${maximum} records`);107 return value.map((candidate, index) => {108 const row = object(candidate, `${label}[${index}]`);109 if (typeof row.title !== 'string' || !row.title.trim() || row.title.length > 500)110 throw new DomainError('INVALID_INPUT', `${label}[${index}].title is invalid`);111 if (typeof row.company !== 'string' || !row.company.trim() || row.company.length > 500)112 throw new DomainError('INVALID_INPUT', `${label}[${index}].company is invalid`);113 for (const [key, item] of Object.entries(row)) {114 if (item !== null && !['string', 'number', 'boolean', 'undefined'].includes(typeof item))115 throw new DomainError('INVALID_INPUT', `${label}[${index}].${key} must be scalar`);116 if (typeof item === 'string' && item.length > 20_000)117 throw new DomainError('RESOURCE_LIMIT_EXCEEDED', `${label}[${index}].${key} is too large`);118 }119 const record = row as JobRecord;120 stableJobKey(record);121 return record;122 });123}124function indexRows(rows: JobRecord[], snapshot: 'baseline' | 'current'): {125 canonical: Map<string, JobRecord>;126 duplicates: Omit<DuplicateGroup, 'comparisonId'>[];127} {128 const grouped = new Map<string, { record: JobRecord; indexes: number[] }>();129 rows.forEach((record, index) => {130 const key = stableJobKey(record);131 const group = grouped.get(key);132 if (group) group.indexes.push(index);133 else grouped.set(key, { record, indexes: [index] });134 });135 return {136 canonical: new Map([...grouped].map(([key, value]) => [key, value.record])),137 duplicates: [...grouped]138 .filter(([, value]) => value.indexes.length > 1)139 .map(([stableKey, value]) => ({140 recordType: 'duplicate-group' as const,141 snapshot,142 stableKey,143 indexes: value.indexes,144 count: value.indexes.length,145 engineVersion: ENGINE_VERSION,146 }))147 .sort((a, b) => a.stableKey.localeCompare(b.stableKey)),148 };149}150export function processInput(value: unknown): RunResult {151 const input = object(value, 'input');152 if (typeof input.comparisonId !== 'string' || !input.comparisonId.trim() || input.comparisonId.length > 120)153 throw new DomainError('INVALID_INPUT', 'comparisonId is required');154 const comparisonId = input.comparisonId;155 const maximum = input.maxRecordsPerSnapshot ?? 1_000;156 if (!Number.isInteger(maximum) || (maximum as number) < 1 || (maximum as number) > 1_000)157 throw new DomainError('INVALID_INPUT', 'maxRecordsPerSnapshot must be an integer from 1 to 1000');158 const baseline = parseRows(input.baseline, 'baseline', maximum as number);159 const current = parseRows(input.current, 'current', maximum as number);160 if (baseline.length + current.length === 0)161 throw new DomainError('INVALID_INPUT', 'at least one snapshot must contain a record');162 if (Buffer.byteLength(canonical({ baseline, current }), 'utf8') > 10_000_000)163 throw new DomainError('RESOURCE_LIMIT_EXCEEDED', 'snapshot input exceeds 10 MB');164 const tracked = input.trackedFields ?? DEFAULT_TRACKED_FIELDS;165 if (!Array.isArray(tracked) || tracked.length < 1 || tracked.length > 30 ||166 tracked.some((field) => typeof field !== 'string' || !field || field.length > 100))167 throw new DomainError('INVALID_INPUT', 'trackedFields must contain 1 to 30 field names');168 if (new Set(tracked).size !== tracked.length)169 throw new DomainError('INVALID_INPUT', 'trackedFields must be unique');170 const oldIndex = indexRows(baseline, 'baseline');171 const newIndex = indexRows(current, 'current');172 const allKeys = [...new Set([...oldIndex.canonical.keys(), ...newIndex.canonical.keys()])].sort();173 let unchanged = 0;174 const changes: Change[] = [];175 for (const key of allKeys) {176 const before = oldIndex.canonical.get(key) ?? null;177 const after = newIndex.canonical.get(key) ?? null;178 let changeType: Change['changeType'];179 let changedFields: Change['changedFields'];180 if (!before) {181 changeType = 'ADDED';182 changedFields = [];183 } else if (!after) {184 changeType = 'REMOVED';185 changedFields = [];186 } else {187 changedFields = (tracked as string[]).flatMap((field) =>188 canonical(before[field]) === canonical(after[field])189 ? []190 : [{ field, before: before[field] ?? null, after: after[field] ?? null }],191 );192 if (!changedFields.length) {193 unchanged += 1;194 continue;195 }196 changeType = 'CHANGED';197 }198 changes.push({199 recordType: 'job-change',200 comparisonId: comparisonId.trim(),201 stableKey: key,202 changeType,203 changedFields,204 baseline: before,205 current: after,206 engineVersion: ENGINE_VERSION,207 });208 }209 const duplicates = [...oldIndex.duplicates, ...newIndex.duplicates].map((item) => ({210 ...item,211 comparisonId: comparisonId.trim(),212 }));213 const inputHash = hash({214 comparisonId: comparisonId.trim(),215 baseline,216 current,217 trackedFields: tracked,218 });219 return {220 inputHash,221 changes,222 duplicates,223 summary: {224 recordType: 'job-summary',225 status: 'SUCCEEDED',226 comparisonId: comparisonId.trim(),227 baselineRecords: baseline.length,228 currentRecords: current.length,229 added: changes.filter((item) => item.changeType === 'ADDED').length,230 removed: changes.filter((item) => item.changeType === 'REMOVED').length,231 changed: changes.filter((item) => item.changeType === 'CHANGED').length,232 unchanged,233 duplicateGroups: duplicates.length,234 inputHash,235 engineVersion: ENGINE_VERSION,236 },237 };238}1import { Actor, log } from 'apify';2
3import { DomainError, processInput } from './domain.js';4import { deliverWithPpe, type PpePlatform, type PpeStore, semanticUnitKey } from './ppe.js';5
6const EVENT_NAME = 'job-snapshots-compared';7
8function platform(): PpePlatform {9 const manager = Actor.getChargingManager();10 return {11 isPayPerEvent: () => manager.getPricingInfo().isPayPerEvent,12 remaining: (eventName) => manager.calculateMaxEventChargeCountWithinLimit(eventName),13 chargedCount: (eventName) => manager.getChargedEventCount(eventName),14 publishCharged: async (item, eventName) => Actor.pushData(item, eventName),15 publishUncharged: async (item) => {16 await Actor.pushData(item);17 },18 };19}20const store: PpeStore = {21 getValue: async (key) => Actor.getValue(key),22 setValue: async (key, value) => Actor.setValue(key, value),23};24
25await Actor.main(async () => {26 try {27 const result = processInput(await Actor.getInput());28 const details = [...result.changes, ...result.duplicates];29 const key = semanticUnitKey('job-comparison', result.inputHash);30 const charging = await deliverWithPpe({31 eventName: EVENT_NAME,32 platform: platform(),33 store,34 units: [{35 idempotencyKey: key,36 billable: true,37 primaryItem: result.summary,38 deliverBeforeCharge: async () => {39 await Actor.setValue('REPORTS', { changes: result.changes, duplicates: result.duplicates });40 if (details.length) await Actor.pushData(details);41 },42 }],43 });44 if (!charging.availableKeys.includes(key)) {45 const limited = {46 status: 'CHARGE_LIMIT_REACHED',47 chargedCount: charging.chargedCount,48 eventChargeLimitReached: charging.eventChargeLimitReached,49 };50 await Actor.setValue('OUTPUT', limited);51 log.warning('No billable work was delivered because the event charge limit was reached', limited);52 return;53 }54 const output = {55 ...result.summary,56 chargedCount: charging.chargedCount,57 requestedChargeCount: charging.requestedCount,58 eventChargeLimitReached: charging.eventChargeLimitReached,59 };60 await Actor.setValue('OUTPUT', output);61 log.info('Job snapshot comparison complete', output);62 } catch (error) {63 if (error instanceof DomainError) log.error(error.message, { code: error.code });64 throw error;65 }66});1import { createHash } from 'node:crypto';2
3export interface ChargeResult {4 eventChargeLimitReached: boolean;5 chargedCount: number;6 chargeableWithinLimit: Record<string, number>;7}8export interface PpePlatform {9 isPayPerEvent(): boolean;10 remaining(eventName: string): number;11 chargedCount(eventName: string): number;12 publishCharged(item: Record<string, unknown>, eventName: string): Promise<ChargeResult>;13 publishUncharged(item: Record<string, unknown>): Promise<void>;14}15export interface PpeStore {16 getValue<T>(key: string): Promise<T | null>;17 setValue(key: string, value: unknown): Promise<void>;18}19export interface PpeUnit {20 idempotencyKey: string;21 billable: boolean;22 primaryItem: Record<string, unknown>;23 deliverBeforeCharge?: () => Promise<void>;24 deliverAfterCharge?: () => Promise<void>;25}26type UnitState = 'delivering' | 'delivered' | 'publishing' | 'charged' | 'complete';27interface LedgerEntry {28 state: UnitState;29 sequence: number;30 requestedCount: number;31 chargedCount: number;32 eventChargeLimitReached: boolean;33 updatedAt: string;34}35interface Ledger {36 version: 1;37 eventName: string;38 nextSequence: number;39 units: Record<string, LedgerEntry>;40}41export interface PpeDiagnostic {42 idempotencyKey: string;43 requestedCount: number;44 chargedCount: number;45 eventChargeLimitReached: boolean;46}47export interface PpeRunResult {48 requestedCount: number;49 chargedCount: number;50 availableKeys: string[];51 duplicateCount: number;52 truncatedCount: number;53 eventChargeLimitReached: boolean;54 diagnostics: PpeDiagnostic[];55}56const LEDGER_KEY = 'PPE_CHARGE_LEDGER_V1';57export const DIAGNOSTICS_KEY = 'PPE_CHARGE_DIAGNOSTICS';58
59export function semanticUnitKey(namespace: string, value: unknown): string {60 return `${namespace}:${createHash('sha256').update(JSON.stringify(value)).digest('hex')}`;61}62function emptyLedger(eventName: string): Ledger {63 return { version: 1, eventName, nextSequence: 1, units: {} };64}65function now(): string {66 return new Date().toISOString();67}68function diagnostic(key: string, entry: LedgerEntry): PpeDiagnostic {69 return {70 idempotencyKey: key,71 requestedCount: entry.requestedCount,72 chargedCount: entry.chargedCount,73 eventChargeLimitReached: entry.eventChargeLimitReached,74 };75}76export function reconcileBatchResult(77 idempotencyKeys: string[],78 result: ChargeResult,79): { chargedKeys: string[]; unchargedKeys: string[] } {80 const charged = Math.max(0, Math.min(idempotencyKeys.length, result.chargedCount));81 return {82 chargedKeys: idempotencyKeys.slice(0, charged),83 unchargedKeys: idempotencyKeys.slice(charged),84 };85}86async function save(store: PpeStore, ledger: Ledger): Promise<void> {87 await store.setValue(LEDGER_KEY, ledger);88}89async function reconcileInterruptedCharge(90 ledger: Ledger,91 platformChargedCount: number,92): Promise<void> {93 const accounted = Object.values(ledger.units).filter((entry) =>94 ['charged', 'complete'].includes(entry.state),95 ).length;96 let unaccounted = Math.max(0, platformChargedCount - accounted);97 const pending = Object.entries(ledger.units)98 .filter(([, entry]) => entry.state === 'publishing')99 .sort(([, left], [, right]) => left.sequence - right.sequence);100 for (const [, entry] of pending) {101 if (unaccounted === 0) break;102 entry.state = 'charged';103 entry.chargedCount = 1;104 entry.updatedAt = now();105 unaccounted -= 1;106 }107}108export async function deliverWithPpe(options: {109 eventName: string;110 units: PpeUnit[];111 platform: PpePlatform;112 store: PpeStore;113}): Promise<PpeRunResult> {114 const { eventName, units, platform, store } = options;115 const stored = await store.getValue<Ledger>(LEDGER_KEY);116 const ledger =117 stored?.version === 1 && stored.eventName === eventName ? stored : emptyLedger(eventName);118 const isPpe = platform.isPayPerEvent();119 if (isPpe) {120 await reconcileInterruptedCharge(ledger, platform.chargedCount(eventName));121 await save(store, ledger);122 }123 const availableKeys: string[] = [];124 const diagnostics: PpeDiagnostic[] = [];125 let chargedCount = 0;126 let duplicateCount = 0;127 let limitReached = false;128 let capacityExhausted = false;129 let truncatedCount = 0;130 for (const unit of units) {131 if (!unit.billable || !isPpe) {132 await unit.deliverBeforeCharge?.();133 await platform.publishUncharged(unit.primaryItem);134 await unit.deliverAfterCharge?.();135 availableKeys.push(unit.idempotencyKey);136 continue;137 }138 let entry = ledger.units[unit.idempotencyKey];139 if (entry?.state === 'complete') {140 duplicateCount += 1;141 availableKeys.push(unit.idempotencyKey);142 diagnostics.push(diagnostic(unit.idempotencyKey, entry));143 continue;144 }145 if (!entry && (capacityExhausted || platform.remaining(eventName) <= 0)) {146 limitReached = true;147 truncatedCount += 1;148 continue;149 }150 if (!entry) {151 entry = {152 state: 'delivering',153 sequence: ledger.nextSequence,154 requestedCount: 1,155 chargedCount: 0,156 eventChargeLimitReached: false,157 updatedAt: now(),158 };159 ledger.nextSequence += 1;160 ledger.units[unit.idempotencyKey] = entry;161 await save(store, ledger);162 }163 if (entry.state === 'delivering') {164 await unit.deliverBeforeCharge?.();165 entry.state = 'delivered';166 entry.updatedAt = now();167 await save(store, ledger);168 }169 if (entry.state === 'delivered' || entry.state === 'publishing') {170 entry.state = 'publishing';171 entry.updatedAt = now();172 await save(store, ledger);173 const result = await platform.publishCharged(unit.primaryItem, eventName);174 entry.requestedCount = 1;175 entry.chargedCount = Math.min(1, Math.max(0, result.chargedCount));176 entry.eventChargeLimitReached = result.eventChargeLimitReached;177 entry.updatedAt = now();178 if (entry.chargedCount !== 1) {179 entry.state = 'delivered';180 await save(store, ledger);181 diagnostics.push(diagnostic(unit.idempotencyKey, entry));182 limitReached = true;183 truncatedCount += 1;184 continue;185 }186 entry.state = 'charged';187 chargedCount += 1;188 await save(store, ledger);189 }190 if (entry.state === 'charged') {191 await unit.deliverAfterCharge?.();192 entry.state = 'complete';193 entry.updatedAt = now();194 await save(store, ledger);195 }196 availableKeys.push(unit.idempotencyKey);197 diagnostics.push(diagnostic(unit.idempotencyKey, entry));198 capacityExhausted = entry.eventChargeLimitReached;199 }200 const result: PpeRunResult = {201 requestedCount: units.filter((unit) => unit.billable).length,202 chargedCount,203 availableKeys,204 duplicateCount,205 truncatedCount,206 eventChargeLimitReached: limitReached,207 diagnostics,208 };209 await store.setValue(DIAGNOSTICS_KEY, result);210 return result;211}1import { describe, expect, it } from 'vitest';2import { DomainError, processInput, stableJobKey } from '../src/domain.js';3
4describe('job feed comparison', () => {5 const input = {6 comparisonId: 'jobs',7 baseline: [8 { sourceId: '1', title: 'Data Analyst', company: 'Northwind', location: 'Berlin', description: 'SQL' },9 { sourceId: '3', title: 'Designer', company: 'Northwind', location: 'Paris' },10 ],11 current: [12 { sourceId: '1', title: 'Senior Data Analyst', company: 'Northwind', location: 'Berlin', description: 'SQL and Python' },13 { sourceId: '2', title: 'Engineer', company: 'Northwind', location: 'Remote' },14 ],15 };16 it('reports additions, removals, and field changes deterministically', () => {17 const first = processInput(input);18 expect(first).toEqual(processInput(input));19 expect(first.summary).toMatchObject({ added: 1, removed: 1, changed: 1, unchanged: 0 });20 expect(first.changes.find((row) => row.stableKey === 'source:1')?.changedFields.map((x) => x.field)).toEqual(['title', 'description']);21 });22 it('uses canonical URL then normalized tuple fallbacks', () => {23 expect(stableJobKey({ title: 'A', company: 'B', url: 'https://WWW.EXAMPLE.com/job/?utm_source=x' })).toBe('url:https://example.com/job');24 expect(stableJobKey({ title: ' Data Analyst ', company: 'NORTHWIND', location: ' Berlin ' })).toBe('tuple:northwind|data analyst|berlin');25 });26 it('reports duplicate stable keys without hiding the canonical comparison', () => {27 const result = processInput({ comparisonId: 'd', baseline: [], current: [28 { sourceId: '1', title: 'A', company: 'B' },29 { sourceId: '1', title: 'A duplicate', company: 'B' },30 ] });31 expect(result.summary).toMatchObject({ added: 1, duplicateGroups: 1 });32 expect(result.duplicates[0]).toMatchObject({ snapshot: 'current', indexes: [0, 1] });33 });34 it('allows custom tracked fields', () => {35 const result = processInput({ ...input, trackedFields: ['location'] });36 expect(result.summary.changed).toBe(0);37 expect(result.summary.unchanged).toBe(1);38 });39 it('fails closed on unusable identity, nested values, invalid URL, and limits', () => {40 expect(() => stableJobKey({ title: 'A', company: 'B' })).toThrow(DomainError);41 expect(() => processInput({ comparisonId: 'x', baseline: [], current: [{ sourceId: '1', title: 'A', company: 'B', nested: {} }] })).toThrow(/scalar/);42 expect(() => processInput({ comparisonId: 'x', baseline: [], current: [{ title: 'A', company: 'B', url: 'bad' }] })).toThrow(/absolute/);43 expect(() => processInput({ comparisonId: 'x', maxRecordsPerSnapshot: 1, baseline: [], current: [44 { sourceId: '1', title: 'A', company: 'B' }, { sourceId: '2', title: 'C', company: 'B' },45 ] })).toThrow(/at most 1/);46 });47});1import { describe, expect, it } from 'vitest';2import { type ChargeResult, deliverWithPpe, type PpePlatform, type PpeStore, semanticUnitKey } from '../src/ppe.js';3
4class MemoryStore implements PpeStore {5 public readonly values = new Map<string, unknown>();6 public async getValue<T>(key: string): Promise<T | null> {7 return (this.values.get(key) as T | undefined) ?? null;8 }9 public async setValue(key: string, value: unknown): Promise<void> {10 this.values.set(key, structuredClone(value));11 }12}13class FakePlatform implements PpePlatform {14 public charges = 0;15 public readonly charged: Record<string, unknown>[] = [];16 public readonly free: Record<string, unknown>[] = [];17 public crashAfterCharge = false;18 public constructor(public capacity: number, private readonly ppe = true) {}19 public isPayPerEvent(): boolean { return this.ppe; }20 public remaining(): number { return Math.max(0, this.capacity - this.charges); }21 public chargedCount(): number { return this.charges; }22 public async publishCharged(item: Record<string, unknown>): Promise<ChargeResult> {23 if (this.remaining() === 0) return { chargedCount: 0, eventChargeLimitReached: true, chargeableWithinLimit: {} };24 this.charged.push(item);25 this.charges += 1;26 if (this.crashAfterCharge) {27 this.crashAfterCharge = false;28 throw new Error('post-charge failure');29 }30 return { chargedCount: 1, eventChargeLimitReached: this.remaining() === 0, chargeableWithinLimit: {} };31 }32 public async publishUncharged(item: Record<string, unknown>): Promise<void> { this.free.push(item); }33}34const unit = (id: string, billable = true) => ({35 idempotencyKey: semanticUnitKey('test', id), billable, primaryItem: { id },36});37describe('PPE coordinator', () => {38 it('charges a completed logical unit once and makes retries idempotent', async () => {39 const platform = new FakePlatform(2);40 const store = new MemoryStore();41 const options = { eventName: 'event', units: [unit('a')], platform, store };42 expect(await deliverWithPpe(options)).toMatchObject({ chargedCount: 1, truncatedCount: 0 });43 expect(await deliverWithPpe(options)).toMatchObject({ chargedCount: 0, duplicateCount: 1 });44 expect(platform.charges).toBe(1);45 });46 it('refuses billable work when event capacity is exhausted', async () => {47 const platform = new FakePlatform(0);48 const result = await deliverWithPpe({ eventName: 'event', units: [unit('a')], platform, store: new MemoryStore() });49 expect(result).toMatchObject({ chargedCount: 0, truncatedCount: 1, eventChargeLimitReached: true });50 expect(platform.charged).toHaveLength(0);51 });52 it('delivers non-billable evidence without charging', async () => {53 const platform = new FakePlatform(0);54 const result = await deliverWithPpe({ eventName: 'event', units: [unit('a', false)], platform, store: new MemoryStore() });55 expect(result.availableKeys).toHaveLength(1);56 expect(platform.free).toEqual([{ id: 'a' }]);57 });58 it('reconciles a crash immediately after the platform charge', async () => {59 const platform = new FakePlatform(2);60 const store = new MemoryStore();61 platform.crashAfterCharge = true;62 await expect(deliverWithPpe({ eventName: 'event', units: [unit('a')], platform, store })).rejects.toThrow('post-charge');63 const retry = await deliverWithPpe({ eventName: 'event', units: [unit('a')], platform, store });64 expect(platform.charges).toBe(1);65 expect(retry.availableKeys).toHaveLength(1);66 });67 it('runs the durable callback before requesting a charge', async () => {68 const order: string[] = [];69 const platform = new FakePlatform(1);70 const original = platform.publishCharged.bind(platform);71 platform.publishCharged = async (item) => { order.push('charge'); return original(item); };72 await deliverWithPpe({73 eventName: 'event', platform, store: new MemoryStore(),74 units: [{ ...unit('a'), deliverBeforeCharge: async () => { order.push('durable'); } }],75 });76 expect(order).toEqual(['durable', 'charge']);77 });78});1import { readFile } from 'node:fs/promises';2import { describe, expect, it } from 'vitest';3import { processInput } from '../src/domain.js';4
5interface Property { prefill?: unknown; }6interface Schema { required?: string[]; properties: Record<string, Property>; }7function materialize(schema: Schema): Record<string, unknown> {8 return Object.fromEntries(Object.entries(schema.properties).flatMap(([name, property]) =>9 Object.hasOwn(property, 'prefill') ? [[name, structuredClone(property.prefill)]] : [],10 ));11}12describe('Store-QA regression contract', () => {13 it('derives the exact safe prefill and truthful repeatable output', async () => {14 const schema = JSON.parse(await readFile(new URL('../.actor/input_schema.json', import.meta.url), 'utf8')) as Schema;15 const manifest = JSON.parse(await readFile(new URL('../actor.manifest.json', import.meta.url), 'utf8')) as {16 store: { storeQaPrefill: Record<string, unknown> };17 };18 const input = materialize(schema);19 expect(schema.required?.every((name) => Object.hasOwn(input, name))).toBe(true);20 expect(input).toEqual(manifest.store.storeQaPrefill);21 expect(JSON.stringify(input)).not.toMatch(/credential|token|requestList/i);22 const first = processInput(input);23 const second = processInput(input);24 expect(first).toEqual(second);25 expect(first.summary).toMatchObject({ baselineRecords: 1, currentRecords: 2, added: 1, changed: 1 });26 });27});{ "extends": "@apify/tsconfig", "compilerOptions": { "module": "NodeNext", "moduleResolution": "NodeNext", "outDir": "dist", "rootDir": ".", "lib": ["ES2023", "DOM"], "strict": true, "noUncheckedIndexedAccess": true, "exactOptionalPropertyTypes": true }, "include": ["src/**/*.ts", "scripts/**/*.ts", "test/**/*.ts"]}