1import { Actor, log } from 'apify';
2import { fetchSource } from './adapters/index.js';
3import { resolveDomain } from './resolve.js';
4import { diffJobs, deriveSignals, hashOf, fnOf } from './diff.js';
5
6const SCHEMA_VERSION = '0.1';
7const TRUNCATION_LIMIT = { smartrecruiters: 100 };
8
9await Actor.init();
10
11const input = await Actor.getInput() || {};
12const sources = (input.sources || []).slice(0, input.maxSources || 200);
13const domains = (input.domains || []).slice(0, input.maxSources || 200);
14const wantSignals = input.deriveSignals !== false;
15const wantEvidence = input.includeEvidence === true;
16const tally = { sources: 0, unchanged: 0, indexed: 0, changes: 0, signals: 0, unresolved: 0, errors: 0 };
17
18
19
20const stateStoreName = `ats-state-${(Actor.getEnv().actorId || 'local').toLowerCase()}`;
21let store;
22try {
23 store = await Actor.openKeyValueStore(stateStoreName);
24} catch (e) {
25 log.warning(`Named state store unavailable (${e.message}). Falling back to the run's default store — every run will be a baseline.`);
26 store = await Actor.openKeyValueStore();
27}
28
29
30if (domains.length) {
31 const map = (await store.getValue('domain-map')) || {};
32 for (const d of domains) {
33 const host = String(d).replace(/^https?:[/][/]/, '').replace(/[/].*$/, '').replace(/^www[.]/, '');
34 if (!map[host]) {
35 const r = await resolveDomain(host);
36 map[host] = r || { ats: null };
37 log.info(r ? `resolved ${host} -> ${r.ats}/${r.company} (${r.method})` : `could not resolve ${host}`);
38 }
39 const m = map[host];
40 if (m && m.ats && !sources.some(x => x.ats === m.ats && x.company === m.company))
41 sources.push({ ats: m.ats, company: m.company, label: host });
42 if (!m || !m.ats)
43 tally.unresolved++,
44 await Actor.pushData({ schema_version: SCHEMA_VERSION, type: 'source_unresolved', type_family: 'meta',
45 domain: host, observed_at: new Date().toISOString(),
46 hint: 'Jobs are on an unsupported ATS or rendered with JavaScript.' });
47 }
48 await store.setValue('domain-map', map);
49}
50
51let stop = false;
52
53const charge = async (eventName, count = 1) => {
54 if (stop || count <= 0) return false;
55 const r = await Actor.charge({ eventName, count });
56 if (r?.eventChargeLimitReached) { stop = true; log.warning(`Charge limit reached at ${eventName}`); }
57 return true;
58};
59
60for (const s of sources) {
61 if (stop) break;
62 const key = `state__${s.ats}__${String(s.company).replace(/[^a-z0-9_-]/gi, '_')}`;
63 const state = (await store.getValue(key)) || { jobs: {}, history: [], functions: {} };
64
65 tally.sources++;
66 await charge('source-checked');
67 const res = await fetchSource(s, state.etag);
68
69 if (res.notModified) {
70 tally.unchanged++;
71 log.info(`${s.ats}/${s.company}: 304 not modified (${res.ms}ms)`);
72 continue;
73 }
74 if (res.error) {
75 tally.errors++;
76 await Actor.pushData({ schema_version: SCHEMA_VERSION, type: 'source_error', type_family: 'meta',
77 source: s, error: res.error, observed_at: new Date().toISOString() });
78 continue;
79 }
80
81 const jobs = res.jobs || [];
82 const limit = TRUNCATION_LIMIT[s.ats];
83 const truncated = Boolean(limit && jobs.length >= limit);
84 const baseline = Object.keys(state.jobs).length === 0;
85 const observed_at = new Date().toISOString();
86 const evidence = { source_url: res.url, http_status: res.status, etag: res.etag,
87 fetch_ms: res.ms, job_count: jobs.length, truncated };
88
89 let events = [];
90 if (baseline) {
91 events = jobs.map(j => ({ type: 'job_indexed', type_family: 'supply', job: j }));
92 } else {
93 events = diffJobs(state.jobs, jobs, { truncated });
94 }
95
96
97 const month = observed_at.slice(0, 7);
98 let mh = state.history.find(h => h.month === month);
99 if (!mh) { mh = { month, posted: 0, total: 0 }; state.history.push(mh); }
100 mh.posted += events.filter(e => e.type === 'job_posted').length;
101 mh.total = jobs.length;
102 state.history = state.history.slice(-24);
103
104 const signals = (!baseline && wantSignals) ? deriveSignals(state.history, events, state) : [];
105 for (const e of events) if (e.type === 'job_posted') { const f = fnOf(e.job); if (f !== 'other') state.functions[f] = observed_at; }
106
107 const emit = async (e, chargeEvent) => {
108 await Actor.pushData({
109 schema_version: SCHEMA_VERSION, observed_at,
110 company: s.company, ats: s.ats, label: s.label || null,
111 ...e,
112 confidence: truncated ? 0.6 : 0.99,
113 evidence: wantEvidence ? evidence : { source_url: evidence.source_url, truncated }
114 });
115 };
116
117 if (baseline) {
118 for (const e of events) await emit(e);
119 tally.indexed += events.length;
120 await charge('job-indexed', events.length);
121 log.info(`${s.ats}/${s.company}: baseline ${events.length} jobs`);
122 } else {
123 for (const e of events) await emit(e);
124 tally.changes += events.length;
125 if (events.length) await charge('change-detected', events.length);
126 for (const e of signals) await emit(e);
127 tally.signals += signals.length;
128 if (signals.length) await charge('signal-derived', signals.length);
129 log.info(`${s.ats}/${s.company}: ${events.length} changes, ${signals.length} signals`);
130 }
131 if (wantEvidence && (events.length || signals.length)) await charge('evidence-full');
132
133 state.etag = res.etag;
134 state.jobs = Object.fromEntries(jobs.map(j => [j.source_job_id, { h: hashOf(j), title: j.title,
135 location: j.location, department: j.department, employment_type: j.employment_type,
136 compensation: j.compensation, url: j.url }]));
137 await store.setValue(key, state);
138}
139
140
141
142
143await Actor.pushData({
144 schema_version: SCHEMA_VERSION,
145 type: 'run_summary',
146 type_family: 'meta',
147 observed_at: new Date().toISOString(),
148 sources_checked: tally.sources,
149 sources_unchanged: tally.unchanged,
150 jobs_indexed: tally.indexed,
151 changes_detected: tally.changes,
152 signals_derived: tally.signals,
153 sources_unresolved: tally.unresolved,
154 source_errors: tally.errors
155});
156log.info(`run summary: ${tally.sources} sources, ${tally.unchanged} unchanged, ${tally.changes} changes, ${tally.signals} signals`);
157
158await Actor.exit();
