Node.js store fulfillment reference

node --version
node --check node/server.mjs

node node/server.mjs serve

node node/server.mjs work

node node/server.mjs inspect

node node/server.mjs export YOUR_ORDER_ID /private/key.json

#!/usr/bin/env node
// Trusted fulfillment receiver/worker, not a payment gateway or a browser client.
// Node.js 22.13+; built-in SQLite, crypto and HTTP. No npm dependencies.
// Private KDP_FULFILLMENT_CONFIG/STATE/SECRET/OUTBOX_KEY and KDP_API_KEY required.
// Run: node server.mjs serve | work | inspect | export YOUR_ORDER_ID /private/key.json
// Expected: safe receipt/work JSON; full keys only in explicit private exports.
import * as fs from 'node:fs';
import path from 'node:path';
import http from 'node:http';
import { createHash, createHmac, timingSafeEqual, randomBytes, createCipheriv, createDecipheriv } from 'node:crypto';
import { DatabaseSync } from 'node:sqlite';

const LIMIT = 262144;
const id = value => typeof value === 'string' && /^[A-Za-z0-9_.:-]{1,190}$/.test(value);
const object = value => value !== null && typeof value === 'object' && !Array.isArray(value);
const sha = value => createHash('sha256').update(value).digest('hex');
const identity = (...values) => sha(values.join('\0'));
const fail = code => { throw new Refusal(code); };
const canonical = value => JSON.stringify(sort(value));
const now = () => Math.floor(Date.now() / 1000);
const delay = ms => new Promise(resolve => setTimeout(resolve, ms));
class Refusal extends Error {}

function sort(value) {
  if (Array.isArray(value)) return value.map(sort);
  if (!object(value)) return value;
  return Object.fromEntries(Object.keys(value).sort().map(key => [key, sort(value[key])]));
}

class Fulfillment {
  constructor() {
    process.umask(0o077);
    const file = process.env.KDP_FULFILLMENT_CONFIG ?? '';
    if (!path.isAbsolute(file)) fail('PRIVATE_ABSOLUTE_CONFIG_REQUIRED');
    this.config = JSON.parse(fs.readFileSync(file, 'utf8'));
    this.directory = process.env.KDP_FULFILLMENT_STATE ?? '';
    if (!path.isAbsolute(this.directory)) fail('PRIVATE_ABSOLUTE_STATE_REQUIRED');
    if (fs.existsSync(this.directory) && fs.lstatSync(this.directory).isSymbolicLink()) fail('INVALID_STATE_DIRECTORY');
    fs.mkdirSync(this.directory, { recursive: true, mode: 0o700 });
    if (fs.statSync(this.directory).mode & 0o077) fail('PRIVATE_STATE_PERMISSIONS_REQUIRED');
    this.secret = process.env.KDP_FULFILLMENT_SECRET ?? '';
    const key = process.env.KDP_FULFILLMENT_OUTBOX_KEY ?? '';
    this.outboxKey = Buffer.from(key, 'base64');
    if (this.secret.length < 32 || this.outboxKey.length !== 32 || this.outboxKey.toString('base64') !== key) fail('PRIVATE_SECRETS_REQUIRED');
    this.environment = this.config.environment;
    if (!['sandbox', 'live'].includes(this.environment)) fail('INVALID_ENVIRONMENT');
    if (this.environment === 'live' && process.env.KDP_CONFIRM_LIVE !== '1') fail('LIVE_CONFIRMATION_REQUIRED');
    for (const field of ['source', 'seller_id', 'product_id']) if (!id(this.config[field])) fail('INVALID_CONFIGURATION');
    if (!object(this.config.offers) || !Object.keys(this.config.offers).length) fail('OFFERS_REQUIRED');
    for (const [offerId, offer] of Object.entries(this.config.offers)) {
      if (!id(offerId) || !id(offer?.plan_id) || !Number.isInteger(offer.renewal_days) || offer.renewal_days < 1 || offer.renewal_days > 3650) fail('INVALID_OFFER_MAPPING');
    }
    this.base = this.config.base_url.replace(/\/+$/, '');
    const url = new URL(this.base);
    const local = this.environment === 'sandbox' && process.env.KDP_ALLOW_LOOPBACK === '1' && url.protocol === 'http:' && url.hostname === '127.0.0.1';
    if (!url.hostname || url.username || url.password || url.search || url.hash || url.pathname !== '/' || (url.protocol !== 'https:' && !local)) fail('INVALID_BASE_URL');
    this.apiKey = process.env.KDP_API_KEY ?? '';
    if (!this.apiKey.startsWith('lmp_' + this.environment + '_')) fail('API_KEY_ENVIRONMENT_MISMATCH');
    const database = path.join(this.directory, 'fulfillment.sqlite');
    if (fs.existsSync(database) && fs.lstatSync(database).isSymbolicLink()) fail('INVALID_STATE_FILE');
    this.db = new DatabaseSync(database);
    fs.chmodSync(database, 0o600);
    this.db.exec(`PRAGMA busy_timeout=3000;
      CREATE TABLE IF NOT EXISTS inbox(event_id TEXT PRIMARY KEY,body_hash TEXT NOT NULL,effect_key TEXT NOT NULL,received_at INTEGER NOT NULL);
      CREATE TABLE IF NOT EXISTS orders(order_key TEXT PRIMARY KEY,offer_id TEXT NOT NULL,email TEXT NOT NULL,license_id TEXT,last_sequence INTEGER NOT NULL DEFAULT 0,held INTEGER NOT NULL DEFAULT 0);
      CREATE TABLE IF NOT EXISTS effects(effect_key TEXT PRIMARY KEY,order_key TEXT NOT NULL,sequence INTEGER NOT NULL,command_hash TEXT NOT NULL,command_json TEXT NOT NULL,state TEXT NOT NULL,request_id TEXT,error_code TEXT,created_at INTEGER NOT NULL,UNIQUE(order_key,sequence));
      CREATE TABLE IF NOT EXISTS outbox(order_key TEXT PRIMARY KEY,license_id TEXT NOT NULL,cipher TEXT NOT NULL,created_at INTEGER NOT NULL);
      CREATE TABLE IF NOT EXISTS binding(id INTEGER PRIMARY KEY CHECK(id=1),config_hash TEXT NOT NULL,outbox_key_hash TEXT NOT NULL);`);
    const configHash = sha(canonical(this.config)), keyHash = sha(this.outboxKey);
    this.run('INSERT OR IGNORE INTO binding(id,config_hash,outbox_key_hash) VALUES(1,?,?)', configHash, keyHash);
    const bound = this.one('SELECT * FROM binding WHERE id=1');
    if (bound.config_hash !== configHash || bound.outbox_key_hash !== keyHash) fail('CONFIGURATION_CHANGED_RECONCILE_FIRST');
  }

  run(sql, ...values) { return this.db.prepare(sql).run(...values); }
  one(sql, ...values) { return this.db.prepare(sql).get(...values); }
  transaction(operation) {
    this.db.exec('BEGIN IMMEDIATE');
    try { const result = operation(); this.db.exec('COMMIT'); return result; }
    catch (error) { this.db.exec('ROLLBACK'); throw error; }
  }

  validate(event) {
    const fields = ['version', 'source', 'seller_id', 'environment', 'event_id', 'effect_id', 'order_id', 'sequence', 'action', 'offer_id', 'customer', 'expected_status', 'expected_expires_at'];
    if (!object(event) || canonical(Object.keys(event).sort()) !== canonical(fields.sort()) || event.version !== 1) fail('INVALID_COMMAND');
    for (const field of ['source', 'seller_id', 'environment']) if (event[field] !== this.config[field]) fail('WRONG_SELLER_OR_ENVIRONMENT');
    for (const field of ['event_id', 'effect_id', 'order_id', 'offer_id']) if (!id(event[field])) fail('INVALID_IDENTIFIER');
    if (!Object.hasOwn(this.config.offers, event.offer_id) || !['issue', 'renew', 'suspend', 'restore', 'revoke', 'record_cancellation'].includes(event.action) || !Number.isInteger(event.sequence) || event.sequence < 1 || event.sequence > 2147483647) fail('UNSUPPORTED_COMMAND');
    const customer = event.customer;
    if (!object(customer) || canonical(Object.keys(customer).sort()) !== '["email","name"]' || typeof customer.email !== 'string' || !/^[^\s@]{1,120}@[^\s@]{1,120}$/.test(customer.email) || customer.email.length > 190 || typeof customer.name !== 'string' || [...customer.name].length > 160) fail('INVALID_CUSTOMER');
    if (event.expected_expires_at !== null && (typeof event.expected_expires_at !== 'string' || event.expected_expires_at.length > 40)) fail('INVALID_EXPECTED_EXPIRY');
    if (event.action === 'issue') {
      if (event.sequence !== 1 || event.expected_status !== null || event.expected_expires_at !== null) fail('INVALID_ISSUANCE_COMMAND');
    } else if (!['active', 'suspended', 'revoked', 'expired', 'pending', 'cancelled'].includes(event.expected_status)) fail('CURRENT_STATE_REQUIRED');
    return event;
  }

  accept(body, signature) {
    const match = /^t=([1-9][0-9]{0,10}),v1=([0-9a-f]{64})$/.exec(signature);
    if (body.length > LIMIT || !match || Math.abs(now() - Number(match[1])) > 300) fail('INVALID_SIGNATURE');
    const actual = createHmac('sha256', this.secret).update(match[1] + '.').update(body).digest();
    if (!timingSafeEqual(actual, Buffer.from(match[2], 'hex'))) fail('INVALID_SIGNATURE');
    const event = this.validate(JSON.parse(new TextDecoder('utf-8', { fatal: true }).decode(body)));
    const orderKey = identity(event.source, event.seller_id, event.environment, event.order_id);
    const effectKey = identity(event.source, event.seller_id, event.environment, event.effect_id);
    const semantic = { ...event }; delete semantic.event_id;
    const json = canonical(semantic), commandHash = sha(json), rawHash = sha(body);
    return this.transaction(() => {
      const previous = this.one('SELECT body_hash FROM inbox WHERE event_id=?', event.event_id);
      if (previous) {
        if (previous.body_hash !== rawHash) fail('EVENT_ID_CONFLICT');
        return { accepted: true, duplicate: true };
      }
      const effect = this.one('SELECT command_hash FROM effects WHERE effect_key=?', effectKey);
      if (effect && effect.command_hash !== commandHash) fail('EFFECT_CONFLICT');
      const order = this.one('SELECT * FROM orders WHERE order_key=?', orderKey);
      if (order && (order.offer_id !== event.offer_id || order.email !== event.customer.email)) fail('ORDER_MAPPING_CONFLICT');
      this.run('INSERT OR IGNORE INTO orders(order_key,offer_id,email) VALUES(?,?,?)', orderKey, event.offer_id, event.customer.email);
      if (!effect) {
        if (this.one('SELECT effect_key FROM effects WHERE order_key=? AND sequence=?', orderKey, event.sequence)) fail('SEQUENCE_CONFLICT');
        this.run("INSERT INTO effects(effect_key,order_key,sequence,command_hash,command_json,state,created_at) VALUES(?,?,?,?,?,'WAITING',?)", effectKey, orderKey, event.sequence, commandHash, json, now());
      }
      this.run('INSERT INTO inbox(event_id,body_hash,effect_key,received_at) VALUES(?,?,?,?)', event.event_id, rawHash, effectKey, now());
      return { accepted: true, duplicate: Boolean(effect) };
    });
  }

  async api(method, route, body = null, idempotency = null) {
    const headers = { Authorization: 'Bearer ' + this.apiKey, Accept: 'application/json', 'Content-Type': 'application/json', 'User-Agent': 'KDPika-Reference-Fulfillment/1' };
    if (idempotency) headers['Idempotency-Key'] = idempotency;
    const response = await fetch(this.base + '/api/v1' + route, {
      method, headers, body: body === null ? undefined : canonical(body), redirect: 'manual', signal: AbortSignal.timeout(15000)
    });
    const reader = response.body.getReader();
    const chunks = []; let size = 0;
    try {
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        size += value.length;
        if (size > 1048576) { await reader.cancel(); fail('UNEXPECTED_API_BODY'); }
        chunks.push(value);
      }
    } finally { reader.releaseLock(); }
    const data = JSON.parse(new TextDecoder('utf-8', { fatal: true }).decode(Buffer.concat(chunks)));
    if (!object(data)) fail('UNEXPECTED_API_BODY');
    return [response.status, data];
  }

  async perform(command, order, effectKey) {
    if (command.action === 'issue') {
      if (order.license_id) fail('ORDER_ALREADY_HAS_LICENSE');
      const [status, result] = await this.api('POST', '/licenses', {
        product_id: this.config.product_id, plan_id: this.config.offers[command.offer_id].plan_id, customer: command.customer
      }, 'store_' + effectKey);
      const item = result.license;
      if (status !== 201 || result.success !== true || !object(item) || typeof item.id !== 'string' || !/^lic_[A-Za-z0-9_-]{1,100}$/.test(item.id) || typeof item.license_key !== 'string' || item.license_key.length < 12 || item.license_key.length > 100 || item.environment !== this.environment) fail('ISSUANCE_RESULT_REQUIRES_REVIEW');
      return [item.id, item.license_key, result.request_id ?? null];
    }
    if (!order.license_id) fail('ISSUE_FIRST');
    const route = '/licenses/' + encodeURIComponent(order.license_id);
    const [status, current] = await this.api('GET', route), item = current.license;
    if (status !== 200 || current.success !== true || !object(item) || item.id !== order.license_id || item.environment !== this.environment || item.status !== command.expected_status || item.expires_at !== command.expected_expires_at) fail('LICENSE_CHANGED_RECONCILE_FIRST');
    if (command.action === 'record_cancellation') return [item.id, null, current.request_id ?? null];
    const body = command.action === 'renew' ? { days: this.config.offers[command.offer_id].renewal_days } : { reason: 'Verified store fulfillment decision' };
    const [changedStatus, changed] = await this.api('POST', route + '/' + command.action, body);
    const expected = { suspend: 'suspended', restore: 'active', revoke: 'revoked', renew: item.status === 'expired' ? 'active' : item.status }[command.action];
    if (changedStatus !== 200 || changed.success !== true || changed.license?.id !== item.id || changed.license.status !== expected || (command.action === 'renew' && typeof changed.license.expires_at !== 'string')) fail('MUTATION_RESULT_REQUIRES_REVIEW');
    return [item.id, null, changed.request_id ?? null];
  }

  async work() {
    // Node core has no flock. An atomic directory lock fails closed after a crash.
    // Never remove a stale lock before confirming its worker is no longer running.
    const lock = path.join(this.directory, 'worker.lockdir');
    try { fs.mkdirSync(lock, { mode: 0o700 }); } catch { fail('WORKER_BUSY_OR_STALE_LOCK'); }
    let processed = 0;
    try {
      while (processed < 25) {
        const claimed = this.transaction(() => {
          const effect = this.one("SELECT e.* FROM effects e JOIN orders o ON o.order_key=e.order_key WHERE e.state='WAITING' AND o.held=0 AND e.sequence=o.last_sequence+1 ORDER BY e.created_at,e.effect_key LIMIT 1");
          if (!effect) return null;
          const order = this.one('SELECT * FROM orders WHERE order_key=?', effect.order_key);
          this.run("UPDATE effects SET state='IN_FLIGHT' WHERE effect_key=?", effect.effect_key);
          this.run('UPDATE orders SET held=1 WHERE order_key=?', effect.order_key);
          return [effect, order];
        });
        if (!claimed) break;
        const [effect, order] = claimed;
        try {
          const [licenseId, key, requestId] = await this.perform(JSON.parse(effect.command_json), order, effect.effect_key);
          if (requestId !== null && (typeof requestId !== 'string' || requestId.length > 190)) fail('UNEXPECTED_REQUEST_ID');
          this.transaction(() => {
            if (key !== null) {
              const nonce = randomBytes(12), cipher = createCipheriv('aes-256-gcm', this.outboxKey, nonce);
              cipher.setAAD(Buffer.from(effect.order_key));
              const bytes = Buffer.concat([nonce, cipher.update(key, 'utf8'), cipher.final(), cipher.getAuthTag()]);
              this.run('INSERT INTO outbox(order_key,license_id,cipher,created_at) VALUES(?,?,?,?)', effect.order_key, licenseId, bytes.toString('base64'), now());
            }
            this.run("UPDATE effects SET state='DONE',request_id=? WHERE effect_key=?", requestId, effect.effect_key);
            this.run('UPDATE orders SET license_id=?,last_sequence=?,held=0 WHERE order_key=?', licenseId, effect.sequence, effect.order_key);
          });
        } catch {
          this.run("UPDATE effects SET state='REVIEW',error_code='RECONCILE_BEFORE_RETRY' WHERE effect_key=?", effect.effect_key);
        }
        processed++; await delay(1100);
      }
    } finally { fs.rmdirSync(lock); }
    return { processed, states: this.inspect() };
  }

  inspect() {
    return Object.fromEntries(this.db.prepare('SELECT state,COUNT(*) AS count FROM effects GROUP BY state').all().map(row => [row.state, row.count]));
  }

  export(orderId, target) {
    if (!id(orderId)) fail('INVALID_ORDER_ID');
    const orderKey = identity(this.config.source, this.config.seller_id, this.environment, orderId);
    const item = this.one('SELECT b.*,o.email FROM outbox b JOIN orders o ON o.order_key=b.order_key WHERE b.order_key=?', orderKey);
    if (!item) fail('NO_SAVED_KEY');
    const parent = path.dirname(target);
    if (!path.isAbsolute(target) || !fs.statSync(parent).isDirectory() || (fs.statSync(parent).mode & 0o077)) fail('PRIVATE_EXPORT_DIRECTORY_REQUIRED');
    const raw = Buffer.from(item.cipher, 'base64');
    if (raw.length < 29) fail('INVALID_OUTBOX');
    const cipher = createDecipheriv('aes-256-gcm', this.outboxKey, raw.subarray(0, 12));
    cipher.setAAD(Buffer.from(orderKey)); cipher.setAuthTag(raw.subarray(-16));
    const key = Buffer.concat([cipher.update(raw.subarray(12, -16)), cipher.final()]).toString('utf8');
    fs.writeFileSync(target, canonical({ license_id: item.license_id, license_key: key, email: item.email }), { flag: 'wx', mode: 0o600 });
    return { exported: true, contains_secret: true };
  }

  serve() {
    const reply = (response, status, value) => {
      const body = canonical(value);
      response.writeHead(status, { 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(body), 'Cache-Control': 'no-store', 'X-Content-Type-Options': 'nosniff', Connection: 'close' });
      response.end(body);
    };
    const server = http.createServer((request, response) => {
      if (request.method !== 'POST' || request.url !== '/store-event') return reply(response, 404, { accepted: false });
      const counts = name => request.rawHeaders.filter((_, index, values) => index % 2 === 0 && values[index].toLowerCase() === name).length;
      const length = request.headers['content-length'];
      if (request.headers['transfer-encoding'] || counts('content-length') !== 1 || !/^\d+$/.test(length ?? '')) return reply(response, 400, { accepted: false });
      if (Number(length) > LIMIT) return reply(response, 413, { accepted: false });
      if (counts('x-fulfillment-signature') !== 1) return reply(response, 401, { accepted: false });
      if ((request.headers['content-type'] ?? '').split(';')[0].trim().toLowerCase() !== 'application/json') return reply(response, 415, { accepted: false });
      const chunks = []; let size = 0;
      request.on('data', chunk => { size += chunk.length; if (size <= LIMIT) chunks.push(chunk); });
      request.on('end', () => {
        if (size > LIMIT) return reply(response, 413, { accepted: false });
        if (size !== Number(length)) return reply(response, 400, { accepted: false });
        try { reply(response, 202, this.accept(Buffer.concat(chunks), request.headers['x-fulfillment-signature'] ?? '')); }
        catch (error) { reply(response, error instanceof Refusal ? (error.message === 'INVALID_SIGNATURE' ? 401 : 409) : (error instanceof SyntaxError || error instanceof TypeError ? 400 : 503), { accepted: false }); }
      });
      request.on('error', () => response.destroy());
    });
    server.requestTimeout = 5000; server.headersTimeout = 5000; server.keepAliveTimeout = 1000;
    server.listen(Number(process.env.KDP_FULFILLMENT_PORT ?? 8112), '127.0.0.1', () => console.log('{"listening":true,"loopback_only":true}'));
  }
}

try {
  const app = new Fulfillment(), args = process.argv.slice(2);
  if (args.length === 1 && args[0] === 'serve') app.serve();
  else if (args.length === 1 && args[0] === 'work') console.log(canonical(await app.work()));
  else if (args.length === 1 && args[0] === 'inspect') console.log(canonical({ states: app.inspect() }));
  else if (args.length === 3 && args[0] === 'export') console.log(canonical(app.export(args[1], args[2])));
  else fail('USE_SERVE_WORK_INSPECT_OR_EXPORT');
} catch {
  console.log('{"completed":false,"error":"REVIEW_PRIVATE_CONFIGURATION_OR_STATE"}');
  process.exitCode = 1;
}