#!/usr/bin/env php
<?php
/**
 * =====================================================================
 *  bin/rx-worker — the thing that actually runs everything
 *
 *  Usage:
 *    rx-worker once            run whatever is due, then exit  (Task Scheduler)
 *    rx-worker run             loop, checking every 30 seconds (service)
 *    rx-worker status          what ran, what is late, what is failing
 *    rx-worker jobs            set up the schedule for every store on this node
 *    rx-worker run-job <name>  force one job now, ignoring its window
 *
 *  Nine engines were built and tested before anything invoked them.
 *  This file is where the schedule meets the code. Every job name in
 *  Scheduler::DEFAULTS must be registered here — tests/test_worker.php
 *  fails the build if one is not, because a job with no handler sits in
 *  the table forever reporting "no handler registered" and nobody looks.
 * =====================================================================
 */
declare(strict_types=1);

$root = dirname(__DIR__);
require $root . '/lib/bootstrap.php';
require $root . '/lib/scheduler.php';
require $root . '/lib/backup.php';
require $root . '/lib/invariants.php';
require $root . '/lib/updater.php';
require $root . '/lib/fileinstaller.php';
require $root . '/lib/messaging.php';
require $root . '/lib/reorder.php';
require $root . '/lib/crosssell.php';
require $root . '/lib/loyalty.php';
require $root . '/lib/publicgate.php';
require $root . '/lib/accounting.php';
require $root . '/lib/returns.php';
require $root . '/lib/providers.php';

/**
 * Resolve a credential reference to a secret.
 *
 * The tables hold `secret://meta/shree-krishna`, never the token. In
 * production this reads the OS keystore or a mounted secrets file; here
 * it reads config, and returns NULL when nothing is configured so that a
 * store without credentials sends nothing rather than erroring.
 */
function rxSecret(string $ref, array $cfg): ?string
{
    $map = $cfg['secrets'] ?? [];
    return isset($map[$ref]) && $map[$ref] !== '' ? (string) $map[$ref] : null;
}

$cfg = require $root . '/config.php';
$db  = new Db($cfg['dsn'], $cfg['db_user'], $cfg['db_pass']);

/**
 * Register every job for one store.
 *
 * Kept in a function so the worker and the test register the SAME set:
 * a handler list that exists only inside a CLI script is a handler list
 * nothing can check.
 */
function registerAll(Scheduler $s, Db $db, string $tenantId, string $storeId, array $cfg): void
{
    $backupDir = ($cfg['data_dir'] ?? dirname(__DIR__) . '/data') . '/backups';
    $key = (string) ($cfg['backup_key'] ?? '');
    $backup = $key !== ''
        ? new BackupEngine($db, $storeId, new LocalBackupStore($backupDir), $key)
        : null;

    $s->register('outbox.drain', function() use ($db, $tenantId, $storeId, $cfg) {
        $w = new MessageWorker($db, $tenantId, $storeId);

        // Everything goes through the dispatcher: template approval, the
        // per-store spend caps and the pricing all live there. A worker
        // that called providers directly would send uncapped.
        $dis = new MessageDispatcher($db, $storeId);
        foreach ($db->all('SELECT DISTINCT provider_code, credential_ref, sender_id
                             FROM message_provider
                            WHERE is_active=1 AND (store_id=? OR store_id IS NULL)',
                          [$storeId]) as $p) {
            $secret = rxSecret((string) $p['credential_ref'], $cfg);
            if ($secret === null) { continue; }   // not configured yet; nothing sends
            if ($p['provider_code'] === 'META_CLOUD') {
                $dis->register(new MetaCloudProvider((string) $p['sender_id'], $secret));
            } else {
                $dis->register(new HttpSmsProvider((string) $p['provider_code'],
                    (string) ($cfg['sms_endpoint'] ?? ''), $secret, (string) $p['sender_id']));
            }
        }
        $w->useDispatcher($dis);

        $out = $w->drain(200);
        $out['spend'] = $dis->spendToday();
        return $out;
    });

    $s->register('sync.push', function() use ($db, $tenantId, $storeId, $cfg) {
        // THIS WAS A STUB THAT COUNTED ROWS.
        //
        // It reported a pending figure and called nothing, so a store
        // node never actually synced: events accumulated in the outbox,
        // masters were never refreshed, and number blocks were never
        // refilled. Principles P3 and P9 both depend on this job, and it
        // did nothing at all. The count it returned even looked healthy.
        $base  = (string) ($cfg['cloud_url'] ?? '');
        $token = rxSecret('node://token', $cfg);
        if ($base === '' || $token === null) {
            // A counter-only shop with no cloud is a supported setup, and
            // the outbox depth is the honest thing to report for it. But
            // say WHY nothing moved rather than returning a number that
            // reads like success.
            return ['skipped' => 'no cloud configured',
                    'outbox'  => (int) $db->val('SELECT COUNT(*) FROM event_outbox', [], 0)];
        }

        $client = new SyncClient($db, new HttpTransport($base, $token), $tenantId, $storeId);
        $up   = $client->push();
        $down = $client->pull();

        return ['pushed'   => $up['sent'] ?? 0,
                'accepted' => $up['accepted'] ?? 0,
                'pulled'   => $down['applied'] ?? 0,
                'outbox'   => $client->outboxDepth()];
    });

    $s->register('backup.run', function() use ($backup) {
        if ($backup === null) { throw new RuntimeException('no backup key configured'); }
        return $backup->run('SCHEDULED');
    });

    $s->register('backup.verify', function() use ($backup, $db, $storeId) {
        if ($backup === null) { throw new RuntimeException('no backup key configured'); }
        $last = $db->val("SELECT id FROM backup_run WHERE store_id=? AND status='UNVERIFIED'
                           ORDER BY taken_at DESC LIMIT 1", [$storeId]);
        if ($last === null) { return ['nothing_to_verify' => true]; }
        // Verification restores into a scratch schema and compares. It is
        // the only thing that turns a written file into a backup.
        return $backup->verify((string) $last, 'rx_verify_' . substr((string) $storeId, -8));
    });

    $s->register('housekeeping', function() use ($db, $tenantId, $storeId) {
        // Four small jobs nobody would ever build a screen for, and which
        // grow without bound if nothing runs them: expired one-time codes
        // that map mobiles to pharmacies, job history, and the supplier
        // lead times the reorder engine learns from.
        $out = [];

        $gate = new PublicGate($db, new RateLimiter($db));
        $out['purged'] = $gate->purge();

        $out['job_history'] = (new Scheduler($db, $storeId))->pruneHistory(30);

        // Lead time is learned from orders received since the last run.
        $re = new ReorderEngine($db, $tenantId, $storeId);
        $learned = 0;
        foreach ($db->all(
            "SELECT id FROM purchase_order
              WHERE store_id=? AND status IN ('RECEIVED','PART_RECEIVED')
                AND closed_at > DATE_SUB(NOW(), INTERVAL 2 DAY)", [$storeId]) as $po) {
            $learned += (int) $re->learnLeadTime((string) $po['id']);
        }
        $out['lead_times_learned'] = $learned;

        return $out;
    });

    $s->register('backup.prune', function() use ($backup) {
        if ($backup === null) { throw new RuntimeException('no backup key configured'); }
        return $backup->prune(7, 4);
    });

    $s->register('accounting.post', function() use ($db, $tenantId, $storeId) {
        $pe = new PostingEngine($db, $tenantId, $storeId);
        // Each posting is idempotent on its own source documents, so a
        // job that runs twice does not double the books.
        $out = [
            'sales'        => $pe->postSales(),
            'purchases'    => $pe->postPurchases(),
            'credit_notes' => $pe->postCreditNotes(),
            'debit_notes'  => $pe->postDebitNotes(),
        ];

        // CLOSING STOCK.
        //
        // Without it a month with heavy buying reads as a loss: the
        // purchases are expensed and the goods on the shelf count for
        // nothing. That was fixed once in the engine and then never run,
        // because nothing called postClosingStock either.
        $ret = new ReturnEngine($db, $tenantId, $storeId);
        $val = $ret->valueStock(date('Y-m-d'));
        $out['closing_stock'] = $pe->postClosingStock(
            date('Y-m-d'), (float) ($val['value_at_cost'] ?? 0));
        return array_map(fn($r) => is_array($r) ? ($r['vouchers'] ?? $r['posted'] ?? 0) : $r, $out);
    });

    $s->register('invariants.check', function() use ($db, $tenantId, $storeId) {
        // Failures land in node_telemetry for the sync daemon to ship, so
        // we find out before the chemist does.
        return (new InvariantChecker($db, $storeId))->runAndReport($tenantId);
    });

    $s->register('crosssell.mine', function() use ($db, $tenantId, $storeId) {
        // BasketRuleMiner had no caller, so no rule was ever learned and
        // suggest() had nothing to suggest. The guards refuse medicine at
        // mining time as well as at suggestion time.
        return (new BasketRuleMiner($db, $tenantId, $storeId))->mine(90);
    });

    $s->register('reorder.run', function() use ($db, $tenantId, $storeId) {
        $re = new ReorderEngine($db, $tenantId, $storeId);
        $run = $re->run();
        $re->draftOrders(date('Y-m-d'));
        return $run;
    });

    $s->register('loyalty.expire', function() use ($db, $tenantId, $storeId) {
        try { $l = new LoyaltyEngine($db, $tenantId, $storeId); }
        catch (LoyaltyException) { return ['skipped' => 'no loyalty scheme for this shop']; }

        // Warn before taking. runExpiry() retires the oldest points
        // first; expiring() is the list the messaging layer uses, and it
        // carries consent so an expiry nudge is gated like any other
        // marketing message.
        $warn = $l->expiring(30);
        return $l->runExpiry() + ['warned' => count($warn)];
    });

    $s->register('refill.queue', function() use ($db, $tenantId, $storeId) {
        $re = new RefillEngine($db, $tenantId, $storeId);
        $re->rebuild();
        return ['queued' => $re->queueDue(new MessageWorker($db, $tenantId, $storeId))];
    });

    $s->register('update.check', function() use ($db, $storeId, $backup, $cfg) {
        if ($backup === null) { return ['skipped' => 'no backup key, so no safe update']; }
        $pub = (string) ($cfg['update_pubkey'] ?? '');
        if ($pub === '') { return ['skipped' => 'no update key configured']; }
        $up = new Updater($db, $storeId, dirname(__DIR__), $backup, $pub);
        $can = $up->canApplyNow();
        // Checking is cheap; applying is not, and it only happens inside
        // the store's own window with the counter quiet.
        return ['can_apply' => $can['ok'], 'reason' => $can['reason']];
    });
}

/** Stores this node is responsible for. */
function activeStores(Db $db): array
{
    return $db->all('SELECT id, tenant_id, code, name FROM store WHERE is_active=1 ORDER BY code');
}

$cmd = $argv[1] ?? 'once';

try {
    switch ($cmd) {

    case 'jobs':
        $n = 0;
        foreach (activeStores($db) as $st) {
            $s = new Scheduler($db, (string) $st['id']);
            $n += $s->ensureJobs();
        }
        echo "Scheduled {$n} new job(s).\n";
        break;

    case 'once':
    case 'run':
        $loop = $cmd === 'run';
        do {
            foreach (activeStores($db) as $st) {
                $s = new Scheduler($db, (string) $st['id']);
                $s->ensureJobs();
                registerAll($s, $db, (string) $st['tenant_id'], (string) $st['id'], $cfg);
                $r = $s->runDue();
                if ($r['ran'] > 0 || $r['failed'] > 0) {
                    printf("%s  %-10s ran %d ok %d failed %d\n",
                           date('H:i:s'), $st['code'], $r['ran'], $r['ok'], $r['failed']);
                    foreach ($r['jobs'] as $job => $outcome) {
                        if ($outcome !== 'ok') { printf("           %-18s %s\n", $job, $outcome); }
                    }
                }
            }
            if ($loop) { sleep(30); }
        } while ($loop);
        break;

    case 'status':
        foreach (activeStores($db) as $st) {
            $s = new Scheduler($db, (string) $st['id']);
            $h = $s->health();
            printf("\n%s — %s\n", $st['code'], $st['name']);
            printf("  %-18s %-22s %-9s %s\n", 'JOB', 'LAST OK', 'FAILURES', 'NEXT DUE');
            foreach ($h['jobs'] as $j) {
                printf("  %-18s %-22s %-9s %s\n", $j['job_name'],
                       $j['last_ok_at'] ?? 'never', $j['consecutive_failures'],
                       $j['next_due_at'] ?? '—');
            }
            foreach ($h['problems'] as $p) { echo "  ! {$p}\n"; }
        }
        echo "\n";
        break;

    case 'run-job':
        $name = $argv[2] ?? '';
        if ($name === '') { fwrite(STDERR, "usage: rx-worker run-job <name>\n"); exit(1); }
        foreach (activeStores($db) as $st) {
            $s = new Scheduler($db, (string) $st['id']);
            registerAll($s, $db, (string) $st['tenant_id'], (string) $st['id'], $cfg);
            // Forced: clears the due time and the window for this one run.
            $db->q("UPDATE scheduled_job SET next_due_at=NULL, window_from_hour=NULL,
                     window_to_hour=NULL WHERE store_id=? AND job_name=?",
                   [$st['id'], $name]);
            $r = $s->runDue();
            printf("%s  %s: %s\n", $st['code'], $name, $r['jobs'][$name] ?? 'not scheduled');
        }
        break;

    default:
        fwrite(STDERR, "usage: rx-worker [once|run|status|jobs|run-job <name>]\n");
        exit(1);
    }

} catch (Throwable $e) {
    // The worker runs unattended. A stack trace on a shop PC helps
    // nobody; the message goes to the log the partner is told to send.
    fwrite(STDERR, "\nThe background worker could not finish.\n\n  "
                 . $e->getMessage() . "\n\nSend this to support.\n\n");
    exit(3);
}
