Example app
A runnable Node 24 receiving server that pulls all five registers safely, with its full source.
This example is a receiving server's scheduled pull. It uses only Node 24's standard library and the public contract: no Kapi code, no npm packages and no database credentials. It is a pattern to copy, not a connector for a particular vendor.
Set it up
- In Works Studio, open Settings → API keys and create a key for the receiving system with
training:read,people:readandcertificates:read. - Store the key in your server's secret manager as
KAPI_WORKS_API_KEY. Keep it out of source code, browser code, command arguments and logs. - Save the source below as
sync.mjs, and make aconnection.jsonwith the API base URL and the company and deployment you expect. Confirm both IDs with the company administrator when you set up the connection; the first training request returns them in itssnapshot. Keep a separate configuration and output folder for each company.
{
"baseUrl": "https://api.kapiworks.com/v1",
"organizationId": "<verified company UUID>",
"deploymentId": "<verified deployment UUID>"
}Run it with the key already in the environment:
node sync.mjs connection.json /private/harbourline/latest.jsonThe example refuses plain HTTP to any host except Kapi's local development API, and it does not follow redirects with the key.
Optional settings in connection.json:
limit: 1 to 200, default 200. Use 2 to exercise paging with a small company.asAt: a timestamp or date for every register. Leave it out to read the current moment. A fixedasAtalways pulls in full.refreshHours: the longest gap between full pulls when nothing has changed. Default 24.from: the start of the people and period training registers. It defaults tohistory_complete_fromfrom the first training pull. Once history is longer than 400 days, set a start within that limit; an unsupported period fails without replacing the previous report.
What it does
It checks for changes first. Before pulling again, it asks /v1/snapshot whether anything changed. If sequence and capture_sequence match the last complete pull, and that pull is less than refreshHours old, it keeps latest.json and stops after one call. Outcomes that depend on the clock, such as not_due_yet becoming overdue, can change without new facts, so a full pull still runs at least daily.
It pulls safely. Pages are fetched one at a time, paced below the per-key limit, and deduplicated by each row's stable ID. Every page has to match the configured company and the pull's first snapshot, and every register has to finish with the number of rows it advertised. Registers can come from different snapshots; use a records pack when you need one snapshot across all of them.
It replaces the report only after a complete pull. It writes a private temporary file and renames it over latest.json, so a failed run keeps the previous report. Repeated pulls replace rows instead of adding duplicates. A lock file next to the output refuses overlapping runs; after a killed process, check it has stopped before you remove a stale lock.
It retries what is worth retrying. Each request times out after 30 seconds. Connection failures and statuses 429, 502, 503 and 504 retry at most five times, with exponential backoff and jitter. Retry-After takes precedence, and a delay of more than five minutes stops the run so your scheduler can try later. 401, 403 and query errors stop at once. The command exits with a non-zero status on failure, so a scheduler can alert someone; it sends no notifications itself.
It summarises without judging. The output holds all five registers, each register's snapshot and page count, the request IDs, the time of the last successful sync and a summary for each employee. Accepted invitations stay separate rows and are not counted again as employees. The summary keeps Kapi's outcomes and times: it does not work out a second training status or turn completion into an exam score. Earlier completions, waivers, exceptions, unknown outcomes and certificates that are not valid are marked for review, and so is no_required_records. The example has no HR roster of its own, so compare its employee IDs with yours to find people who are missing from Kapi.
Certificates need a check address. The API gives check_code; build the address as https://www.kapiworks.com/verify/<check_code>.
The example keeps whole registers in memory. For much larger imports, load the pages into staging tables in your database and swap them in one transaction, keeping the same rule: every page first, then publish.
Source
#!/usr/bin/env node
// Standalone receiving-system example. Uses only Node's standard library and the public /v1 contract.
import http from 'node:http';
import https from 'node:https';
import {mkdir, open, readFile, rename, rm, writeFile} from 'node:fs/promises';
import {dirname, resolve} from 'node:path';
import {pathToFileURL} from 'node:url';
import {setTimeout as delay} from 'node:timers/promises';
import {randomUUID} from 'node:crypto';
const completed = new Set(['completed_on_time', 'completed_late']);
const outcomes = new Set([...completed, 'completed_before_requirement', 'not_due_yet', 'overdue', 'missed',
'excused', 'waived', 'optional_not_done']);
const tally = (rows, field) => rows.reduce((counts, row) => {
counts[row[field]] = (counts[row[field]] ?? 0) + 1;
return counts;
}, Object.create(null));
function readJson(url, key) {
return new Promise((resolveResponse, reject) => {
const local = url.hostname === 'api.localhost';
const request = (url.protocol === 'https:' ? https : http).request(url, {
method: 'GET', headers: {Authorization: `Bearer ${key}`, Accept: 'application/json'},
// Equivalent to the guide's curl --resolve, without changing the computer's hosts file.
...(local ? {lookup: (_host, _options, done) => done(null, [{address: '127.0.0.1', family: 4}])} : {}),
}, response => {
const chunks = [];
let size = 0;
response.on('data', chunk => {
size += chunk.length;
if (size > 3 * 1024 * 1024) request.destroy(new Error('Response exceeds the documented 3 MiB limit'));
else chunks.push(chunk);
});
response.on('error', () => {clearTimeout(timer); reject(new Error('API response was interrupted'));});
response.on('end', () => {
clearTimeout(timer);
try {
resolveResponse({status: response.statusCode, headers: response.headers,
body: JSON.parse(Buffer.concat(chunks).toString('utf8'))});
} catch { reject(new Error('API returned invalid JSON')); }
});
});
const timer = setTimeout(() => request.destroy(new Error('API request timed out')), 30_000);
request.on('error', () => {clearTimeout(timer); reject(new Error('API connection failed or timed out'));});
request.end();
});
}
export function summarize(registers) {
const {training, people, certificates, courses} = registers;
const releases = new Map(courses.data.map(row => [row.course_release_id, row.course_title]));
const employees = people.data.filter(row => row.entry_type === 'employee').map(person => {
const records = training.data.filter(row => row.learner_profile_id === person.learner_profile_id);
const issued = certificates.data.filter(row => row.learner_profile_id === person.learner_profile_id);
const review = new Set(person.no_required_records ? ['no_required_records'] : []);
for (const row of records) {
if (!outcomes.has(row.outcome)) review.add('unknown_outcome');
if (['waived', 'excused', 'completed_before_requirement'].includes(row.outcome)) review.add(row.outcome);
}
for (const row of issued) if (row.status !== 'valid') review.add(`${row.status}_certificate`);
return {personId: person.learner_profile_id, name: person.person_name, access: person.access_state,
completed: records.filter(row => completed.has(row.outcome)).length,
outcomes: tally(records, 'outcome'), review: [...review],
training: records.map(row => ({recordId: row.record_id, courseId: row.course_id,
course: releases.get(row.course_release_id) ?? row.course_id, requirement: row.requirement,
outcome: row.outcome, dueAt: row.due_at ?? null, completedAt: row.completed_at ?? null}))};
});
return {rosterCoverage: 'Kapi employees only; reconcile against your authoritative employee roster.',
training: tally(training.data, 'outcome'), certificates: tally(certificates.data, 'status'),
invitations: people.data.filter(row => row.entry_type === 'invitation').length, employees};
}
export async function pullCompany(config, {key = process.env.KAPI_WORKS_API_KEY, pause = delay, previous = null} = {}) {
const base = new URL(config.baseUrl);
if (base.protocol !== 'https:' && !(base.protocol === 'http:' && base.hostname === 'api.localhost')) {
throw new Error('Use HTTPS; only api.localhost permits HTTP for local development');
}
if (base.username || base.password || base.search || base.hash || !/^\/v1\/?$/.test(base.pathname)) {
throw new Error('baseUrl must be the API origin followed by /v1, without credentials or query parameters');
}
if (!key || /[\r\n]/.test(key)) throw new Error('Set KAPI_WORKS_API_KEY in the receiving server environment');
const uuid = /^[a-f0-9]{8}-(?:[a-f0-9]{4}-){3}[a-f0-9]{12}$/i;
if (!uuid.test(config.organizationId) || !uuid.test(config.deploymentId)) {
throw new Error('Configure the expected organizationId and deploymentId for this connection');
}
const limit = config.limit ?? 200;
if (!Number.isInteger(limit) || limit < 1 || limit > 200) throw new Error('limit must be 1 to 200');
const calls = [];
async function get(path, parameters) {
const url = new URL(base.href.replace(/\/$/, '') + path);
url.search = new URLSearchParams(parameters).toString();
for (let attempt = 0; attempt <= 5; attempt++) {
// One sequential caller stays below the per-key limit. 429 still handles other clients in the company.
await pause(1100);
let response;
try { response = await readJson(url, key); }
catch (error) {
if (attempt === 5) throw error;
await pause(1000 * 2 ** attempt + Math.random() * 250);
continue;
}
const {status, headers, body} = response;
const requestId = String(headers['x-request-id'] ?? 'unavailable').replace(/[^a-zA-Z0-9-]/g, '').slice(0, 80);
calls.push({endpoint: path, status, requestId});
if (status === 200) return body;
const code = /^[a-z_]+$/.test(body.code) ? body.code : 'unexpected_response';
const error = new Error(`HTTP ${status} ${code}; request ID ${requestId}`);
if (attempt === 5 || ![429, 502, 503, 504].includes(status)) throw error;
const retryAfter = headers['retry-after'];
const seconds = retryAfter && /^\d+$/.test(retryAfter) ? Number(retryAfter)
: Math.max(0, (Date.parse(retryAfter ?? '') - Date.now()) / 1000) || 0;
if (seconds > 300) throw new Error(`${error.message}; reschedule after Retry-After (${seconds}s)`);
await pause(Math.max(1000 * 2 ** attempt, seconds * 1000) + Math.random() * 250);
}
}
async function pull(path, id, parameters = {}) {
const rows = new Map(), cursors = new Set();
let cursor = null, snapshot, total, signature, pages = 0;
do {
const body = await get(path, cursor === null ? {limit, ...parameters} : {limit, cursor});
const s = body.snapshot;
if (!s || s.organization_id !== config.organizationId || s.deployment_id !== config.deploymentId) {
throw new Error(`${path}: company or deployment does not match this connection`);
}
if (!Number.isSafeInteger(s.sequence) || !Number.isSafeInteger(s.capture_sequence) ||
typeof s.chain_hash !== 'string' || !Array.isArray(body.data) || !Number.isSafeInteger(body.total) || body.total < 0 ||
!(body.next_cursor === null || (typeof body.next_cursor === 'string' && body.next_cursor.length > 0))) {
throw new Error(`${path}: invalid page envelope`);
}
const nextSignature = JSON.stringify([s.sequence, s.capture_sequence, s.chain_hash, s.as_at, s.from,
s.known_at, s.logic?.version, s.logic?.date, s.history_complete_from]);
if (pages && (signature !== nextSignature || total !== body.total)) throw new Error(`${path}: snapshot changed during paging`);
snapshot = s; signature = nextSignature; total = body.total; pages++;
for (const row of body.data) {
if (typeof row[id] !== 'string' || !row[id] ||
(row.organization_id && row.organization_id !== config.organizationId) ||
(row.deployment_id && row.deployment_id !== config.deploymentId)) throw new Error(`${path}: invalid row identity`);
rows.set(row[id], row);
}
cursor = body.next_cursor;
if (cursor !== null && cursors.has(cursor)) throw new Error(`${path}: repeated cursor`);
cursors.add(cursor);
} while (cursor !== null);
if (rows.size !== total) throw new Error(`${path}: received ${rows.size} unique rows, expected ${total}`);
return {snapshot, pages, total, data: [...rows.values()]};
}
// One call answers "has anything changed?". Facts can still age without a change (not_due_yet becomes overdue),
// so a full pull still runs at least every refreshHours (default 24).
const last = previous?.registers?.training?.snapshot;
if (last && !config.asAt && Date.now() - Date.parse(previous.syncedAt) < (config.refreshHours ?? 24) * 3_600_000) {
const {snapshot} = await get('/snapshot', {});
if (snapshot?.organization_id === config.organizationId && snapshot?.deployment_id === config.deploymentId &&
snapshot.sequence === last.sequence && snapshot.capture_sequence === last.capture_sequence) {
return {...previous, unchanged: true, checkedAt: new Date().toISOString(), calls};
}
}
const training = await pull('/training-records', 'record_id', config.asAt ? {as_at: config.asAt} : {});
const span = {from: config.from ?? training.snapshot.history_complete_from, to: training.snapshot.as_at};
const registers = {training,
period: await pull('/training-records/period', 'record_id', span),
people: await pull('/people', 'person_id', span),
certificates: await pull('/certificates', 'certificate_id', {as_at: training.snapshot.as_at}),
courses: await pull('/courses', 'row_id')};
return {schema: 'kapi.works.example-sync/v1', syncedAt: new Date().toISOString(),
organizationId: config.organizationId, deploymentId: config.deploymentId,
registers, summary: summarize(registers), calls};
}
export async function syncToFile(config, output, options) {
await mkdir(dirname(output), {recursive: true, mode: 0o700});
const lock = await open(output + '.lock', 'wx', 0o600);
const temporary = output + '.' + randomUUID() + '.tmp';
try {
const previous = await readFile(output, 'utf8').then(JSON.parse).catch(() => null);
const report = await pullCompany(config, {...options, previous});
// An unchanged snapshot keeps the last complete pull exactly as it was saved.
if (report.unchanged) return report;
await writeFile(temporary, JSON.stringify(report, null, 2) + '\n', {mode: 0o600, flag: 'wx'});
await rename(temporary, output);
return report;
} finally {
await rm(temporary, {force: true});
await lock.close();
await rm(output + '.lock', {force: true});
}
}
if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
try {
const [configPath, outputPath] = process.argv.slice(2);
if (!configPath || !outputPath) throw new Error('Usage: node sync.mjs connection.json /private/company/latest.json');
const report = await syncToFile(JSON.parse(await readFile(configPath, 'utf8')), resolve(outputPath));
console.log(JSON.stringify({syncedAt: report.syncedAt, unchanged: Boolean(report.unchanged), employees: report.summary.employees.length,
invitations: report.summary.invitations, training: report.summary.training, certificates: report.summary.certificates}));
} catch (error) { console.error(error.message); process.exitCode = 1; }
}