worker.js

const { Worker } = require("bullmq");
const Redis = require("ioredis");
const { createHmac, createHash } = require("crypto");
const { Pool } = require("pg");
const { lookup } = require("dns/promises");
const { isIP } = require("net");
const logger = require("./worker-log.js");

const { createOperatorAlerter, noopAlerter } = require("./worker-alert.js");
const { buildDailyDigest, sendDailyDigest } = require("./worker-digest.js");

// Operator alerts are wired up in startWorker(); until then (and in tests
// that require this module) they are a no-op.
let alerts = noopAlerter;
let opsRedis = null;
const startedAt = Date.now();
const DIGEST_HOUR_UTC = Number(process.env.DIGEST_HOUR_UTC || 8);
// Loop failures repeat every tick while the cause persists; one message per
// half hour per loop is enough to act on.
const LOOP_ALERT_WINDOW_SECONDS = 1800;

function alertLoopError(event, err) {
  const message = err instanceof Error ? err.message : String(err);
  return alerts.send(event, `${event}\n${message}`, { windowSeconds: LOOP_ALERT_WINDOW_SECONDS });
}

// Dead-man's switch. Every loop records its last error-free completion; once
// a minute the worker pings HEARTBEAT_URL only if every loop is fresh and
// Redis answers. A hung or erroring loop withholds the ping, and the external
// monitor expecting it is what pages the operator: the one alert that still
// works when nothing on this host can send one. Tolerances are a little over
// two periods so a single slow tick does not trip it.
const HEARTBEAT_INTERVAL_MS = Number(process.env.HEARTBEAT_INTERVAL_MS || 60_000);
const LOOP_TOLERANCE_MS = {
  webhook_delivery: 30_000,
  activation_expiry: 3 * 60_000,
  hygiene: 2 * 60 * 60_000,
  restriction: 2 * 60 * 60_000,
  deletion: 2 * 60 * 60_000,
  digest: 2 * 60 * 60_000,
};
const lastLoopOkAt = new Map();

function markLoopOk(loop, at = Date.now()) {
  if (loop) lastLoopOkAt.set(loop, at);
}

// A loop that has never completed counts from process start, so a loop that
// fails from the very first tick is reported rather than ignored.
function _resetLoopStateForTests() {
  lastLoopOkAt.clear();
}

function staleLoops(now = Date.now(), since = startedAt) {
  return Object.entries(LOOP_TOLERANCE_MS)
    .filter(([loop, tolerance]) => now - (lastLoopOkAt.get(loop) ?? since) > tolerance)
    .map(([loop]) => loop);
}

// GET the ping URL; best effort and never throws, because a monitoring
// vendor being down must not become a worker problem.
async function pingHeartbeat(url, fetchImpl = fetch) {
  if (!url) return false;
  try {
    const res = await fetchImpl(url, { signal: AbortSignal.timeout(5000) });
    if (!res.ok) {
      logger.warn("heartbeat_ping_rejected", { status: res.status });
      return false;
    }
    return true;
  } catch (err) {
    logger.warn("heartbeat_ping_failed", { error: err });
    return false;
  }
}

async function redisAnswers(redis, timeoutMs = 2000) {
  if (!redis) return false;
  try {
    const pong = await Promise.race([
      redis.ping(),
      new Promise((_, reject) => setTimeout(() => reject(new Error("redis ping timeout")), timeoutMs)),
    ]);
    return pong === "PONG";
  } catch {
    return false;
  }
}

async function heartbeatTick() {
  const stale = staleLoops();
  const redisOk = await redisAnswers(opsRedis);
  if (stale.length > 0 || !redisOk) {
    logger.warn("heartbeat_withheld", { stale, redisOk });
    await alerts.send(
      "worker_unhealthy",
      `Worker unhealthy\nstale loops: ${stale.join(", ") || "none"}\nredis: ${redisOk ? "ok" : "unreachable"}`,
      { windowSeconds: LOOP_ALERT_WINDOW_SECONDS },
    );
    return;
  }
  await pingHeartbeat(process.env.HEARTBEAT_URL);
}

// The tunnel is a separate failure domain with its own external probe, so it
// does not gate the worker's ping; it only raises an alert while the host can
// still send one.
async function checkTunnelReady(url = process.env.CLOUDFLARED_READY_URL, fetchImpl = fetch) {
  if (!url) return;
  let detail;
  try {
    const res = await fetchImpl(url, { signal: AbortSignal.timeout(3000) });
    if (res.ok) return;
    detail = `status ${res.status}`;
  } catch (err) {
    detail = err instanceof Error ? err.message : String(err);
  }
  logger.warn("cloudflared_not_ready", { detail });
  await alerts.send("cloudflared_not_ready", `Cloudflare tunnel not ready\n${detail}`, {
    windowSeconds: LOOP_ALERT_WINDOW_SECONDS,
  });
}

// Graceful-shutdown state. On SIGTERM/SIGINT we stop scheduling new work,
// let the in-flight delivery batch drain (bounded), then close the pool and
// redis so a deploy does not sever connections mid-write.
const GRACEFUL_SHUTDOWN_TIMEOUT_MS = Number(process.env.GRACEFUL_SHUTDOWN_TIMEOUT_MS || 10000);
let isShuttingDown = false;
let inFlightBatches = 0;
const intervalIds = [];
let bullWorker = null;
let bullConnection = null;

// SSRF guard. Webhook URLs are user-supplied; refuse anything pointing at
// localhost, RFC1918, link-local, cloud metadata, etc. Resolve-then-fetch
// leaves a small DNS-rebinding window which we accept for now because the
// outbound traffic is bounded by a 5-second connection timeout.
const BLOCKED_HOSTNAMES = new Set([
  "localhost",
  "metadata.google.internal",
  "metadata",
  "ip6-localhost",
  "ip6-loopback",
]);

function isBlockedIpv4(ip) {
  const parts = ip.split(".").map(Number);
  if (parts.length !== 4 || parts.some(p => !Number.isInteger(p) || p < 0 || p > 255)) {
    return true;
  }
  const [a, b] = parts;
  if (a === 10) return true;
  if (a === 127) return true;
  if (a === 0) return true;
  if (a === 169 && b === 254) return true;
  if (a === 172 && b >= 16 && b <= 31) return true;
  if (a === 192 && b === 168) return true;
  if (a === 100 && b >= 64 && b <= 127) return true;
  if (a >= 224) return true;
  return false;
}

function isBlockedIpv6(ip) {
  const lower = ip.toLowerCase();
  if (lower === "::" || lower === "::1") return true;
  if (lower.startsWith("fe80:") || lower.startsWith("fc") || lower.startsWith("fd")) return true;
  if (lower.startsWith("::ffff:")) {
    return isBlockedIpv4(lower.slice("::ffff:".length));
  }
  return false;
}

function isLoopbackHostname(hostname) {
  if (hostname === "localhost" || hostname === "ip6-localhost" || hostname === "ip6-loopback") {
    return true;
  }
  if (hostname === "::1") return true;
  return isIP(hostname) === 4 && hostname.startsWith("127.");
}

async function assertWebhookUrlIsSafe(rawUrl) {
  let url;
  try {
    url = new URL(rawUrl);
  } catch {
    throw new Error("webhook url is invalid");
  }
  if (url.protocol !== "https:" && url.protocol !== "http:") {
    throw new Error("webhook url protocol is not supported");
  }
  const hostname = url.hostname.toLowerCase();
  // Outside production, allow delivery to loopback so local receivers
  // (and the delivery test fixture) work. The deployed worker runs with
  // NODE_ENV=production, where loopback stays blocked as an SSRF target.
  if (process.env.NODE_ENV !== "production" && isLoopbackHostname(hostname)) {
    return;
  }
  if (BLOCKED_HOSTNAMES.has(hostname)) {
    throw new Error("webhook hostname is not allowed");
  }
  const ipVersion = isIP(hostname);
  if (ipVersion === 4) {
    if (isBlockedIpv4(hostname)) throw new Error("webhook ip is not allowed");
    return;
  }
  if (ipVersion === 6) {
    if (isBlockedIpv6(hostname)) throw new Error("webhook ip is not allowed");
    return;
  }
  const records = await lookup(hostname, { all: true });
  for (const record of records) {
    const blocked = record.family === 6 ? isBlockedIpv6(record.address) : isBlockedIpv4(record.address);
    if (blocked) {
      throw new Error("webhook url resolves to a blocked address");
    }
  }
}

const databaseUrl = process.env.DATABASE_URL;

// Webhook delivery loop — polls the database for pending deliveries,
// signs each request with the per-endpoint plaintext secret, POSTs to
// the receiver, and updates status with backoff on failure. Kept in
// the same process as the telegram worker to share container resources;
// see plan note that crash-safety relies on the DB next_attempt_at
// rather than BullMQ persistence.

// Small explicit pool: the worker only polls + sweeps, so it does not need
// the default 10 connections. Keeping it small protects the Postgres
// connection budget when multiple worker replicas run.
const pool = new Pool({ connectionString: databaseUrl, max: 5 });

// An idle client dropped by Postgres emits 'error' on the pool; unlistened,
// node escalates that to an unhandled 'error' event and kills the worker
// mid-sweep. With the listener the pool just discards the client.
pool.on("error", err => logger.error("db_pool_error", { error: err }));

// After attempt N fails, wait this many seconds before attempt N+1.
// Index 0 is unused; we look up by attempt_count (1-indexed).
const RETRY_DELAYS_SECONDS = [
  0,        // never used
  60,       // after 1st fail → 1 min
  5 * 60,   // after 2nd → 5 min
  15 * 60,  // after 3rd → 15 min
  60 * 60,  // after 4th → 1 h
  4 * 60 * 60,
  12 * 60 * 60,
  24 * 60 * 60,
];
const MAX_ATTEMPTS = RETRY_DELAYS_SECONDS.length - 1;
const DELIVERY_TIMEOUT_MS = 5000;
const RESPONSE_BODY_LIMIT = 4096;
// Disable an endpoint after this many deliveries fail in a row (a
// success resets the count), so a dead receiver stops accruing retries.
const AUTO_DISABLE_THRESHOLD = 5;
// How many due deliveries one tick reserves.
const CLAIM_BATCH_SIZE = 10;

function signPayload(secret, timestamp, body) {
  return createHmac("sha256", secret)
    .update(`${timestamp}.${body}`)
    .digest("hex");
}

async function deliverOne(row) {
  const timestamp = Math.floor(Date.now() / 1000);
  const body = JSON.stringify({
    id: row.public_id,
    type: row.event_type,
    created: timestamp,
    data: row.payload,
  });
  const signature = signPayload(row.secret, timestamp, body);

  const controller = new AbortController();
  const abortTimer = setTimeout(() => controller.abort(), DELIVERY_TIMEOUT_MS);

  let responseStatus = null;
  let responseBody = null;
  let lastError = null;

  try {
    await assertWebhookUrlIsSafe(row.url);
    const response = await fetch(row.url, {
      method: "POST",
      headers: {
        "content-type": "application/json",
        "x-bottleneck-timestamp": String(timestamp),
        "x-bottleneck-signature": signature,
        "x-bottleneck-event": row.event_type,
        "x-bottleneck-delivery": row.public_id,
      },
      body,
      signal: controller.signal,
      redirect: "manual",
    });
    responseStatus = response.status;
    if (response.status >= 300 && response.status < 400) {
      lastError = `unexpected redirect: ${response.status}`;
      responseBody = "";
    } else {
      const text = await response.text();
      responseBody = text.slice(0, RESPONSE_BODY_LIMIT);
      if (!response.ok) {
        lastError = `HTTP ${response.status}`;
      }
    }
  } catch (err) {
    lastError = err instanceof Error ? err.message : String(err);
  } finally {
    clearTimeout(abortTimer);
  }

  const succeeded = responseStatus !== null && responseStatus >= 200 && responseStatus < 300;
  const nextAttempt = row.attempt_count + 1;

  if (succeeded) {
    await pool.query(
      `update webhook_deliveries
          set status = 'delivered',
              attempt_count = $2,
              delivered_at = now(),
              response_status = $3,
              response_body = $4,
              last_error = null,
              next_attempt_at = null
        where id = $1`,
      [row.id, nextAttempt, responseStatus, responseBody],
    );
    await pool.query(
      `update webhook_endpoints set consecutive_failures = 0 where id = $1`,
      [row.webhook_endpoint_id],
    );
    return;
  }

  if (nextAttempt >= MAX_ATTEMPTS) {
    await pool.query(
      `update webhook_deliveries
          set status = 'failed',
              attempt_count = $2,
              response_status = $3,
              response_body = $4,
              last_error = $5,
              next_attempt_at = null
        where id = $1`,
      [row.id, nextAttempt, responseStatus, responseBody, lastError],
    );
    await disableEndpointIfFailing(row);
    return;
  }

  const delaySeconds = RETRY_DELAYS_SECONDS[nextAttempt];
  await pool.query(
    `update webhook_deliveries
        set status = 'pending',
            attempt_count = $2,
            response_status = $3,
            response_body = $4,
            last_error = $5,
            next_attempt_at = now() + ($6 || ' seconds')::interval
      where id = $1`,
    [row.id, nextAttempt, responseStatus, responseBody, lastError, String(delaySeconds)],
  );
}

// A delivery just exhausted its retries. Count it against the endpoint
// and, once enough have failed back-to-back, disable the endpoint so a
// dead receiver stops generating retry load. The update is scoped to
// active endpoints so the transition (and its security event) fire
// exactly once.
async function disableEndpointIfFailing(row) {
  const { rows } = await pool.query(
    `update webhook_endpoints
        set consecutive_failures = consecutive_failures + 1,
            status = case when consecutive_failures + 1 >= $2 then 'disabled' else status end,
            disabled_at = case when consecutive_failures + 1 >= $2 then now() else disabled_at end
      where id = $1 and status = 'active'
      returning status, consecutive_failures`,
    [row.webhook_endpoint_id, AUTO_DISABLE_THRESHOLD],
  );
  const endpoint = rows[0];
  if (endpoint && endpoint.status === "disabled") {
    // Give the backlog a terminal state now that nothing will deliver it.
    // Left pending it would never be claimed (the claim requires an active
    // endpoint) and never purged (the hygiene sweep only deletes terminal
    // rows). An operator replays what matters from the admin page after
    // re-enabling the endpoint.
    await pool.query(
      `update webhook_deliveries
          set status = 'cancelled', next_attempt_at = null
        where webhook_endpoint_id = $1 and status = 'pending'`,
      [row.webhook_endpoint_id],
    );
    await pool.query(
      `insert into security_events (event_type, result, metadata)
       values ('webhook_endpoint_auto_disabled', 'disabled', $1::jsonb)`,
      [
        JSON.stringify({
          webhookEndpointId: row.webhook_endpoint_id,
          deliveryPublicId: row.public_id,
          consecutiveFailures: endpoint.consecutive_failures,
        }),
      ],
    );
    logger.warn("webhook_endpoint_auto_disabled", {
      endpointId: row.webhook_endpoint_id,
      consecutiveFailures: endpoint.consecutive_failures,
    });
    await alerts.send(
      `webhook_disabled:${row.webhook_endpoint_id}`,
      `Webhook endpoint auto-disabled\nendpoint #${row.webhook_endpoint_id} after ${endpoint.consecutive_failures} consecutive failures`,
    );
  }
}

async function processWebhookBatch() {
  // Atomic claim: a single UPDATE reserves up to CLAIM_BATCH_SIZE pending
  // rows by pushing next_attempt_at five minutes into the future, then
  // returns the payload + endpoint metadata needed to deliver. Two
  // concurrent ticks (we run setInterval at 1s; a slow batch can
  // overlap with the next tick) cannot reserve the same row because
  // SKIP LOCKED gives each one a disjoint set. If a worker crashes
  // mid-delivery the row's next_attempt_at unblocks naturally five
  // minutes later — which is far longer than any HTTP attempt's
  // 5-second AbortController timeout.
  //
  // The endpoint must be filtered inside the locked subselect. Filtering it
  // only in the outer UPDATE lets LIMIT spend the window on rows belonging
  // to a disabled endpoint and then decline to update them: they stay
  // pending with next_attempt_at in the past, so once CLAIM_BATCH_SIZE of
  // them are due they fill every subsequent window and no active endpoint is
  // ever served again. `for update of c` locks the delivery rows only;
  // locking the joined endpoint row too would make SKIP LOCKED skip every
  // delivery of an endpoint that is being updated concurrently.
  try {
    const { rows } = await pool.query(
      `update webhook_deliveries d
          set next_attempt_at = now() + interval '5 minutes'
         from webhook_endpoints e
        where d.id in (
                select c.id
                  from webhook_deliveries c
                  join webhook_endpoints ce on ce.id = c.webhook_endpoint_id
                 where c.status = 'pending'
                   and c.next_attempt_at is not null
                   and c.next_attempt_at <= now()
                   and ce.status = 'active'
                 order by c.next_attempt_at
                 limit $1
                 for update of c skip locked
              )
          and e.id = d.webhook_endpoint_id
        returning d.id, d.public_id, d.event_type, d.payload, d.attempt_count,
                  d.webhook_endpoint_id, e.url, e.secret`,
      [CLAIM_BATCH_SIZE],
    );

    for (const row of rows) {
      try {
        await deliverOne(row);
      } catch (err) {
        logger.error("webhook_delivery_threw", { deliveryId: row.public_id, error: err });
      }
    }
  } catch (err) {
    logger.error("webhook_batch_error", { error: err });
    await alertLoopError("webhook_batch_error", err);
    return false;
  }
}

// Hourly DB hygiene sweep. These tables grow without bound otherwise:
//   - oauth_client_assertion_jtis: every private_key_jwt exchange inserts one
//   - registration_requests: holds password hashes for incomplete signups
//   - webhook_deliveries: failed and delivered rows accumulate forever
// The deletes are scoped so an in-flight retry / fresh signup / recent
// delivery is never affected.
async function sweepHygiene() {
  try {
    await pool.query(
      `delete from oauth_client_assertion_jtis
        where expires_at < now() - interval '1 hour'`,
    );
    await pool.query(
      `delete from registration_requests
        where (
                status in ('pending', 'expired', 'cancelled')
                and expires_at < now() - interval '1 day'
              )
           or (
                status in ('verified', 'completed')
                and expires_at < now() - interval '30 days'
              )`,
    );
    await pool.query(
      `delete from webhook_deliveries
        where status in ('delivered', 'failed', 'cancelled')
          and created_at < now() - interval '30 days'`,
    );
    // Deliveries stranded on an endpoint that was disabled outside the
    // owner and auto-disable paths (a direct SQL disable per the runbook, or
    // an enqueue that raced the disable). Without this they sit pending
    // forever: unclaimable and never old enough to purge.
    await pool.query(
      `update webhook_deliveries d
          set status = 'cancelled', next_attempt_at = null
         from webhook_endpoints e
        where e.id = d.webhook_endpoint_id
          and e.status <> 'active'
          and d.status = 'pending'`,
    );
    // Sessions: revoked or expired for over an hour. Active sessions (not
    // revoked, not past expiry) never match.
    await pool.query(
      `delete from sessions
        where (revoked_at is not null or expires_at < now())
          and coalesce(revoked_at, expires_at) < now() - interval '1 hour'`,
    );
    // OAuth authorization codes: consumed or expired for over an hour.
    await pool.query(
      `delete from oauth_authorization_codes
        where (consumed_at is not null or expires_at < now())
          and coalesce(consumed_at, expires_at) < now() - interval '1 hour'`,
    );
    // OAuth access tokens: revoked or expired for over an hour.
    await pool.query(
      `delete from oauth_access_tokens
        where (revoked_at is not null or expires_at < now())
          and coalesce(revoked_at, expires_at) < now() - interval '1 hour'`,
    );
    // OAuth refresh tokens: keep a 30-day grace past revocation/expiry so the
    // rotation chain remains available for reuse detection within a token's
    // own lifetime, then purge (covers rotated/replaced rows too, so the chain
    // does not grow without bound).
    await pool.query(
      `delete from oauth_refresh_tokens
        where (revoked_at is not null or expires_at < now())
          and coalesce(revoked_at, expires_at) < now() - interval '30 days'`,
    );
    // Pushed authorization requests: expired or consumed for over an hour.
    await pool.query(
      `delete from oauth_pushed_requests
        where expires_at < now() - interval '1 hour'
           or (consumed_at is not null and consumed_at < now() - interval '1 hour')`,
    );
    // Device codes: terminal and expired for over an hour.
    await pool.query(
      `delete from oauth_device_codes
        where status in ('expired', 'consumed', 'denied')
          and expires_at < now() - interval '1 hour'`,
    );
    // Telegram login challenges: terminal and expired for over an hour.
    await pool.query(
      `delete from telegram_login_challenges
        where status in ('verified', 'expired', 'cancelled')
          and expires_at < now() - interval '1 hour'`,
    );
    // Activation requests: terminal, kept 7 days so an integrator can still
    // poll the final status after the user decided or it lapsed.
    await pool.query(
      `delete from activation_requests
        where status in ('approved', 'denied', 'expired', 'cancelled')
          and expires_at < now() - interval '7 days'`,
    );
    // Profile change requests: pending ones hold a proposed username/email
    // until Telegram approval; drop dead pending rows after a day and terminal
    // rows after a week.
    await pool.query(
      `delete from profile_change_requests
        where (status = 'pending' and expires_at < now() - interval '1 day')
           or (status in ('approved', 'denied', 'expired', 'cancelled')
               and expires_at < now() - interval '7 days')`,
    );
    // Security events: 90-day audit retention, bounded per sweep so a large
    // backlog drains over several hourly runs instead of one long transaction.
    await pool.query(
      `delete from security_events
        where id in (
          select id from security_events
           where created_at < now() - interval '90 days'
           order by id
           limit 5000
        )`,
    );
  } catch (err) {
    logger.error("hygiene_sweep_error", { error: err });
    await alertLoopError("hygiene_sweep_error", err);
    return false;
  }
}

// Transition pending activations past their expiry to 'expired' and
// enqueue an activation.expired webhook for every active endpoint
// subscribed to it. One statement so the state change and the delivery
// enqueue commit together; gen_random_uuid keeps the delivery public_id
// in the same shape the app's enqueue path produces. Apps that rely on
// webhooks instead of polling otherwise never learn an activation lapsed.
async function sweepExpiredActivations() {
  try {
    const { rows } = await pool.query(
      `with expired as (
         update activation_requests
            set status = 'expired'
          where status = 'pending' and expires_at <= now()
         returning public_id, external_app_id, scopes
       ),
       app_info as (
         select x.public_id, x.external_app_id, x.scopes, a.public_id as app_public_id
           from expired x
           join external_apps a on a.id = x.external_app_id
       ),
       enqueued as (
         insert into webhook_deliveries (public_id, webhook_endpoint_id, event_type, payload, next_attempt_at)
         select 'whd_' || replace(gen_random_uuid()::text, '-', ''),
                ep.id,
                'activation.expired',
                jsonb_build_object(
                  'id', ai.public_id,
                  'status', 'expired',
                  'appId', ai.app_public_id,
                  'scopes', to_jsonb(ai.scopes),
                  'expiredAt', now()
                ),
                now()
           from app_info ai
           join webhook_endpoints ep
             on ep.external_app_id = ai.external_app_id
            and ep.status = 'active'
            and 'activation.expired' = any(ep.event_types)
         returning id
       )
       select (select count(*) from expired)::int as expired_count,
              (select count(*) from enqueued)::int as enqueued_count`,
    );
    const expiredCount = rows[0] ? rows[0].expired_count : 0;
    const enqueuedCount = rows[0] ? rows[0].enqueued_count : 0;
    if (expiredCount > 0) {
      logger.info("activations_expired", { expiredCount, enqueuedCount });
    }
  } catch (err) {
    logger.error("activation_expiry_sweep_error", { error: err });
    await alertLoopError("activation_expiry_sweep_error", err);
    return false;
  }
}

// Fail the boot, not the first job: the worker cannot run without its
// database and bot token, so assert them up front instead of crashing deep
// in the first delivery. The web app validates its own larger secret set
// in lib/server/config.ts; the worker container only receives and needs
// these two (see docker-compose worker service), so it does not assert the
// OIDC/CSRF/Turnstile secrets it never uses.
function validateConfig() {
  const required = ["DATABASE_URL", "TELEGRAM_BOT_TOKEN"];
  const missing = required.filter(name => !process.env[name]);
  if (missing.length > 0) {
    throw new Error(`missing required environment variables: ${missing.join(", ")}`);
  }
}

// A restriction whose security thread has had no activity from the restricted
// user for the inactivity window is auto-closed and the account is banned. Only
// acts on status='active' rows and transitions them first, so a mid-run crash
// (the loop re-runs hourly) cannot double-ban. Idempotent.
async function sweepRestrictedInactive() {
  try {
    const days = Number(process.env.RESTRICTION_INACTIVITY_DAYS || 60);
    const banned = await pool.query(
      `with stale as (
         select r.id, r.user_id, r.trigger_code
           from user_restrictions r
          where r.status = 'active'
            and coalesce(r.last_user_activity_at, r.created_at)
                < now() - make_interval(days => $1)
          limit 200
       ), closed as (
         update user_restrictions r
            set status = 'closed', lifted_at = now()
           from stale s
          where r.id = s.id
       )
       update users u
          set status = 'banned', restricted = false, restricted_at = null, updated_at = now()
         from stale s
        where u.id = s.user_id
        returning u.id as user_id, s.trigger_code`,
      [days],
    );
    for (const row of banned.rows) {
      await pool.query(
        `update sessions set revoked_at = now() where user_id = $1 and revoked_at is null`,
        [row.user_id],
      );
      await pool.query(
        `insert into security_events (user_id, event_type, result, ip, user_agent, country, metadata)
         values ($1, 'restriction_auto_ban', 'ok', '', 'worker', '', $2::jsonb)`,
        [row.user_id, JSON.stringify({ triggerCode: row.trigger_code, days })],
      );
    }
    if (banned.rows.length > 0) {
      logger.info("restriction_auto_ban", { count: banned.rows.length });
    }
  } catch (err) {
    logger.error("restriction_sweep_error", { error: err });
    await alertLoopError("restriction_sweep_error", err);
    return false;
  }
}

// Purge accounts whose grace-period soft delete has elapsed. Signing in before
// the deadline clears deletion_requested_at (createUserSession), so anything
// still set past the window is a confirmed deletion. We reserve the Telegram
// identity in `bans` first (so a recreated account stays blocked even after the
// row is gone), then delete the user, which cascades to their data.
async function sweepPendingDeletions() {
  const days = Number(process.env.DELETION_GRACE_DAYS || 30);
  let due;
  try {
    due = await pool.query(
      `select id
         from users
        where deletion_requested_at is not null
          and deletion_requested_at < now() - make_interval(days => $1)
        limit 200`,
      [days],
    );
  } catch (err) {
    logger.error("deletion_sweep_error", { error: err });
    await alertLoopError("deletion_sweep_error", err);
    return false;
  }

  let purged = 0;
  for (const { id } of due.rows) {
    // Each purge is its own transaction with a re-check under a row lock.
    // Signing in during the window clears deletion_requested_at
    // (createUserSession), so a user who cancelled between the scan above and
    // this delete must NOT be purged - the FOR UPDATE re-check enforces that
    // atomically.
    const client = await pool.connect();
    try {
      await client.query("begin");
      const locked = await client.query(
        `select public_id, telegram_id
           from users
          where id = $1
            and deletion_requested_at is not null
            and deletion_requested_at < now() - make_interval(days => $2)
          for update`,
        [id, days],
      );
      if (locked.rowCount === 0) {
        await client.query("rollback");
        continue;
      }
      const { public_id: publicId, telegram_id: telegramId } = locked.rows[0];

      // "Delete my account" implies erasure: strip PII (IP / user-agent /
      // country / metadata) from this user's audit rows now, while user_id
      // still resolves - the delete below SET NULLs them, so they'd otherwise
      // survive the 90-day retention window carrying personal data.
      await client.query(
        `update security_events
            set ip = '', user_agent = '', country = '', metadata = '{}'::jsonb
          where user_id = $1`,
        [id],
      );

      if (telegramId) {
        const valueHash = createHash("sha256").update(telegramId).digest("hex");
        await client.query(
          `insert into bans (kind, user_id, value_hash, reason, created_by_user_id)
           values ('telegram_id', null, $1, 'account deleted', null)
           on conflict (kind, value_hash) where revoked_at is null and value_hash is not null
             do nothing`,
          [valueHash],
        );
      }
      await client.query(
        `insert into security_events (user_id, event_type, result, ip, user_agent, country, metadata)
         values ($1, 'account_purged', 'ok', '', 'worker', '', $2::jsonb)`,
        [id, JSON.stringify({ publicId, days })],
      );
      await client.query(`delete from users where id = $1`, [id]);
      await client.query("commit");
      purged += 1;
    } catch (err) {
      try {
        await client.query("rollback");
      } catch {
        // ignore rollback failure; the connection is released below
      }
      logger.error("deletion_sweep_error", { error: err, userId: id });
    } finally {
      client.release();
    }
  }

  if (purged > 0) {
    logger.info("account_purge", { count: purged });
  }
}

// Wires up the Redis-backed telegram worker and the DB poll loops. Kept
// behind a require.main guard so the delivery functions can be imported
// by tests without opening a Redis connection or starting the timers.
// Runs a periodic job while tracking it as in-flight so graceful shutdown can
// wait for it to finish. Skips scheduling once shutdown has begun.
function runBatch(fn, errorEvent, loop) {
  if (isShuttingDown) return;
  inFlightBatches += 1;
  fn()
    .then(result => {
      // Loop bodies swallow their own errors and resolve false; a throw
      // lands in catch. Either way the loop is not marked fresh.
      if (result !== false) markLoopOk(loop);
    })
    .catch(err => {
      logger.error(errorEvent, { error: err });
      return alertLoopError(errorEvent, err);
    })
    .finally(() => {
      inFlightBatches -= 1;
    });
}

async function shutdownGracefully(signal) {
  if (isShuttingDown) return;
  isShuttingDown = true;
  logger.info("worker_shutdown_start", { signal, inFlightBatches });
  for (const id of intervalIds) clearInterval(id);
  try {
    if (bullWorker) await bullWorker.close();
  } catch (err) {
    logger.error("worker_close_failed", { error: err });
  }
  const deadline = Date.now() + GRACEFUL_SHUTDOWN_TIMEOUT_MS;
  while (inFlightBatches > 0 && Date.now() < deadline) {
    await new Promise(resolve => setTimeout(resolve, 100));
  }
  if (inFlightBatches > 0) {
    logger.warn("worker_shutdown_timeout", { inFlightBatches });
  }
  try {
    await pool.end();
  } catch (err) {
    logger.error("pool_end_failed", { error: err });
  }
  try {
    if (bullConnection) await bullConnection.quit();
    if (opsRedis) await opsRedis.quit();
  } catch (err) {
    logger.error("redis_quit_failed", { error: err });
  }
  logger.info("worker_shutdown_complete", { signal });
  process.exit(0);
}

function startWorker() {
  validateConfig();
  const botToken = process.env.TELEGRAM_BOT_TOKEN;

  const redisUrl = process.env.REDIS_URL || "redis://localhost:6379";
  bullConnection = new Redis(redisUrl, { maxRetriesPerRequest: null });
  // ioredis emits 'error' on every failed reconnect attempt; without a
  // listener node treats the first one as fatal.
  bullConnection.on("error", err => logger.error("redis_error", { error: err }));

  // Alerts and the digest use their own bounded connection: bullConnection
  // runs with maxRetriesPerRequest: null, so a command on it would block for
  // the whole of a Redis outage, which is the moment alerts matter most.
  opsRedis = new Redis(redisUrl, { maxRetriesPerRequest: 1, enableOfflineQueue: false });
  opsRedis.on("error", err => logger.error("redis_error", { error: err, connection: "ops" }));
  alerts = createOperatorAlerter({
    redis: opsRedis,
    chatId: process.env.ALERT_TELEGRAM_CHAT_ID || process.env.BEARER_ADMIN_TELEGRAM_ID,
    token: botToken,
    enabled: process.env.NODE_ENV === "production",
    log: logger,
  });

  bullWorker = new Worker("telegram-notifications", async (job) => {
    if (job.name === "send") {
      logger.debug("telegram_job_send", { jobId: job.id });
      const res = await fetch(`https://api.telegram.org/bot${botToken}/sendMessage`, {
        method: "POST",
        headers: { "content-type": "application/json" },
        body: JSON.stringify(job.data),
      });

      if (!res.ok) {
        const errText = await res.text();
        logger.error("telegram_api_error", { status: res.status, body: errText });
        throw new Error(`Telegram API error: ${res.status}`);
      }
    }
  }, { connection: bullConnection });

  bullWorker.on("error", err => logger.error("telegram_worker_error", { error: err }));
  bullWorker.on("completed", job => logger.debug("telegram_job_completed", { jobId: job.id }));
  bullWorker.on("failed", (job, err) => logger.error("telegram_job_failed", { jobId: job?.id, error: err }));
  logger.info("telegram_worker_started");

  // Poll roughly once a second. A crashed worker leaves pending rows in
  // the DB; they are picked up on next start.
  intervalIds.push(setInterval(() => runBatch(processWebhookBatch, "webhook_loop_error", "webhook_delivery"), 1000));
  logger.info("webhook_delivery_loop_started");

  intervalIds.push(setInterval(() => runBatch(sweepHygiene, "hygiene_loop_error", "hygiene"), 60 * 60 * 1000));
  // Run once at startup so the first sweep doesn't wait an hour.
  runBatch(sweepHygiene, "initial_hygiene_sweep_error", "hygiene");
  logger.info("hygiene_sweep_started");

  // Activations carry a short TTL (minutes), so sweep every minute to
  // fire activation.expired close to the actual lapse.
  intervalIds.push(setInterval(() => runBatch(sweepExpiredActivations, "activation_sweep_loop_error", "activation_expiry"), 60 * 1000));
  runBatch(sweepExpiredActivations, "initial_activation_sweep_error", "activation_expiry");
  logger.info("activation_expiry_sweep_started");

  // Restricted accounts inactive for the threshold get auto-banned (the case is
  // closed). Hourly is plenty for a 60-day clock.
  intervalIds.push(setInterval(() => runBatch(sweepRestrictedInactive, "restriction_sweep_loop_error", "restriction"), 60 * 60 * 1000));
  runBatch(sweepRestrictedInactive, "initial_restriction_sweep_error", "restriction");
  logger.info("restriction_sweep_started");

  // Soft-deleted accounts past their grace window get purged. Hourly is plenty
  // for a 30-day clock.
  intervalIds.push(setInterval(() => runBatch(sweepPendingDeletions, "deletion_sweep_loop_error", "deletion"), 60 * 60 * 1000));
  runBatch(sweepPendingDeletions, "initial_deletion_sweep_error", "deletion");
  logger.info("deletion_sweep_started");

  // Hourly tick; the hour gate and the 36h NX window inside make it one
  // send per UTC day. Running once at start covers a restart during the
  // digest hour.
  // sendDailyDigest resolves false outside the digest hour; that is a skip,
  // not a failure, so the loop still counts as fresh.
  const digest = async () => {
    await sendDailyDigest({ pool, alerts, redis: opsRedis, hourUtc: DIGEST_HOUR_UTC, startedAt });
  };
  intervalIds.push(setInterval(() => runBatch(digest, "digest_loop_error", "digest"), 60 * 60 * 1000));
  runBatch(digest, "initial_digest_error", "digest");
  logger.info("daily_digest_started", { hourUtc: DIGEST_HOUR_UTC });

  intervalIds.push(setInterval(() => runBatch(heartbeatTick, "heartbeat_error"), HEARTBEAT_INTERVAL_MS));
  intervalIds.push(setInterval(() => runBatch(checkTunnelReady, "tunnel_check_error"), HEARTBEAT_INTERVAL_MS));
  logger.info("heartbeat_started", {
    intervalMs: HEARTBEAT_INTERVAL_MS,
    heartbeatUrl: Boolean(process.env.HEARTBEAT_URL),
    tunnelCheck: Boolean(process.env.CLOUDFLARED_READY_URL),
  });

  process.on("SIGTERM", () => shutdownGracefully("SIGTERM"));
  process.on("SIGINT", () => shutdownGracefully("SIGINT"));

  // Node already terminates on either of these; the handlers exist so the
  // reason is a structured line rather than a bare stack, and so the exit is
  // non-zero and Docker's restart policy takes over. No graceful drain: after
  // an unhandled throw the loop state is unknown, and shutdownGracefully
  // would run against it.
  process.on("unhandledRejection", err => {
    logger.error("unhandled_rejection", { error: err });
    process.exit(1);
  });
  process.on("uncaughtException", err => {
    logger.error("uncaught_exception", { error: err });
    process.exit(1);
  });
}

// `node worker.js --digest` builds and sends the digest once, bypassing the
// hour gate and the daily window, then exits: the runbook's "is the alert
// channel alive" check.
async function runDigestOnce() {
  const redisUrl = process.env.REDIS_URL || "redis://localhost:6379";
  const redis = new Redis(redisUrl, { maxRetriesPerRequest: 1, enableOfflineQueue: false });
  redis.on("error", err => logger.error("redis_error", { error: err, connection: "ops" }));
  const sender = createOperatorAlerter({
    redis,
    chatId: process.env.ALERT_TELEGRAM_CHAT_ID || process.env.BEARER_ADMIN_TELEGRAM_ID,
    token: process.env.TELEGRAM_BOT_TOKEN,
    enabled: true,
    log: logger,
  });
  try {
    const text = await buildDailyDigest(pool, { redis, startedAt });
    const sent = await sender.send(`digest:manual:${Date.now()}`, text, { windowSeconds: 1 });
    logger.info("digest_sent_once", { sent });
    process.stdout.write(text + "\n");
  } finally {
    await pool.end();
    redis.disconnect();
  }
}

if (require.main === module) {
  if (process.argv.includes("--digest")) {
    runDigestOnce().catch(err => {
      logger.error("digest_once_failed", { error: err });
      process.exit(1);
    });
  } else {
    startWorker();
  }
}

module.exports = {
  deliverOne,
  processWebhookBatch,
  sweepHygiene,
  sweepExpiredActivations,
  signPayload,
  RETRY_DELAYS_SECONDS,
  MAX_ATTEMPTS,
  AUTO_DISABLE_THRESHOLD,
  CLAIM_BATCH_SIZE,
  LOOP_TOLERANCE_MS,
  markLoopOk,
  staleLoops,
  _resetLoopStateForTests,
  pingHeartbeat,
  checkTunnelReady,
};